Realtime Compute for Apache Flink:Python İşleri Geliştirme

Table API/SQL ile DataStream API Arasında Seçim Yapma

PyFlink iki geliştirme yaklaşımını destekler: Table API/SQL ve DataStream API. Table API/SQL aşağıdaki nedenlerle önerilir:

• Daha iyi performans: Table API/SQL'in optimize edilmiş yürütme planı tamamen JVM içinde çalışır. Buna karşılık DataStream API, JVM ile Python işlemi arasında satır bazında serileştirme ve serileştirmeyi kaldırma gerektirir; bu da önemli bir performans yükü getirir.

• Daha kapsamlı özellikler: Table API/SQL, konektörler, veri biçimleri ve pencere işlevleri için daha eksiksiz destek sunar ve SQL işleriyle aynı konektör ekosistemini paylaşır.

• Topluluk tavsiyesi: Apache Flink topluluğu, PyFlink için birincil geliştirme yönü olarak Table API/SQL'e öncelik verir.

DataStream API'yi yalnızca SQL ile ifade edilemeyen karmaşık özel mantık senaryolarında kullanın.

Geliştirme Referansları

Flink iş kodunu yerel olarak geliştirmek için aşağıdaki belgelere başvurabilirsiniz. Geliştirme tamamlandıktan sonra kodu Flink geliştirme konsoluna yükleyip işi dağıtın.

• Apache Flink V1.20 iş kodu geliştirmesi için bkz. Flink Python API Geliştirme Kılavuzu.

• Apache Flink kodlama sırasında karşılaşılan sorunlar ve çözümleri için bkz. SSS.

Proje Yapısı

Python işleri için önerilen proje yapısı aşağıdaki gibidir:

my-flink-python-project/
├── my_job.py                # Main job file.
├── udfs.py                  # User-defined functions (optional)
├── requirements.txt         # Third-party Python dependencies (optional)
└── config.properties        # Configuration file (optional)

Bağımlılık Yönetimi

Python işlerinde özel Python sanal ortamlarının, üçüncü taraf Python paketlerinin, JAR dosyalarının ve veri dosyalarının nasıl kullanılacağı hakkında bilgi için bkz. Python bağımlılıklarını kullanma.

Kullanıcı Tanımlı İşlevler (UDF'ler)

Aşağıda, hassas dize verilerini maskeleyen bir Python UDSF geliştirme örneği verilmiştir:

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())def mask_phone(phone: str):"""Phone number masking: retains the first 3 and last 4 digits, replacing the middle with ****"""if phone is None or len(phone) != 11:return phone.
    return phone[:3] + '****' + phone[7:]

Bu UDF'yi bir SQL işinde şu şekilde kullanın:

CREATE TEMPORARY FUNCTION mask_phone AS 'udfs.mask_phone' LANGUAGE PYTHON;INSERT INTO sink_table
SELECT name, mask_phone(phone) AS masked_phone
FROM source_table;

UDF'lerin nasıl kaydedileceği, güncelleneceği ve silineceği hakkında bilgi için bkz. UDF'leri yönetme.

Konektörleri Kullanma

Flink tarafından desteklenen konektörlerin listesi için bkz. Desteklenen konektörler. Bir konektörü kullanmak için aşağıdaki adımları uygulayın:

1. Realtime Compute konsoluna giriş yapın.

2. Hedef çalışma alanının Actions sütununda Console seçeneğine tıklayın.

3. Sol gezinme bölmesinde Artifacts seçeneğine tıklayın.

4. Upload Artifact seçeneğine tıklayın ve hedef konektörün Python paketini seçerek yükleyin.

   Kendi geliştirdiğiniz bir konektörü veya Flink tarafından sağlanan bir konektörü yükleyebilirsiniz. Flink tarafından sağlanan konektörlerin resmi Python paketlerini indirmek için bkz. Konektör listesi.

5. O&M > Deployments sayfasında Create Deployment > Python Deployment seçeneğine tıklayın. Additional Dependencies alanı için hedef konektörün Python paketini seçin, diğer parametreleri yapılandırın ve işi dağıtın.

6. Dağıtılan işin adına tıklayın. Configuration sekmesinde, Parameters bölümündeki Edit seçeneğine tıklayın. Other Configuration alanında Python konektör paketlerinin konum bilgilerini ekleyin.

   İşiniz birden çok konektör Python paketine bağımlıysa, örneğin connector-1.jar ve connector-2.jar adlı iki paket varsa yapılandırma aşağıdaki gibidir.

   pipeline.classpaths: 'file:///flink/usrlib/connector-1.jar;file:///flink/usrlib/connector-2.jar'

7. Yerleşik konektörleri, veri biçimlerini ve katalogları (yalnızca VVR 11.2 ve sonraki sürümler) kullanmak için işin Additional Configuration altındaki Running Parameters bölümüne yapılandırmayı ekleyin. Örnek:

   ## Multiple connectors used.
   pipeline.used-builtin-connectors: kafka;sls
   ## Multiple data formats for data transmission.
   pipeline.used-builtin-formats: avro;parquet
   ## Multiple previously created catalogs used.
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

Konektörlerin nasıl kullanılacağına ilişkin ayrıntılar için bkz. Tam örnek kod.

İş Hata Ayıklama

import logging

@udf(result_type=DataTypes.BIGINT())
def add(i, j):
  logging.info("hello world")
  return i + j

Yerel Hata Ayıklama

Realtime Compute for Apache Flink varsayılan olarak İnternet erişimine sahip olmadığından kodunuz yerel ortamda çevrimiçi veri kaynaklarına doğrudan bağlanamayabilir. Yerel hata ayıklama için aşağıdaki yaklaşımları öneririz:

• Birim testi: İşlev mantığının doğru olduğundan emin olmak için UDF'lerde bağımsız birim testleri gerçekleştirin.

• Yerel yürütme: Girişleri simüle etmek için yerel veri kaynaklarını (dosyalar veya bellek içi veriler gibi) kullanın ve işleme mantığını doğrulamak için işi yerel olarak çalıştırın. Örnek:

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()# Use a local data source for testing.
  ds = env.from_collection([('Alice', 1), ('Bob', 2), ('Alice', 3)])
  ds.key_by(lambda x: x[0]).sum(1).print()
  env.execute("local_test")

• Uzaktan hata ayıklama: Çevrimiçi veri kaynaklarına bağlanarak hata ayıklamak için bkz. Konektörlerle işleri yerel olarak çalıştırma ve hata ayıklama.

İş Dağıtımı

Python işi geliştirmesi tamamlandıktan sonra işi dağıtmak üzere Realtime Compute konsoluna yükleyin. Aşağıdaki adımları uygulayın:

1. Realtime Compute konsoluna giriş yapın ve hedef çalışma alanına gidin.

2. Sol gezinme bölmesinde Artifacts seçeneğine tıklayın ve Python iş dosyasını (.py veya .zip) yükleyin. Üçüncü taraf bağımlılıklar veya yapılandırma dosyaları varsa bunları da yükleyin.

3. O&M > Deployments sayfasında Create Deployment > Python Deployment seçeneğine tıklayın ve dağıtım bilgilerini doldurun.

   Parameter	Description
   Python File Path	Yüklenen Python iş dosyasını seçin.
   Entry Module	İş dosyası bir .py dosyasıysa bu alan gerekli değildir. İş dosyası bir .zip dosyasıysa giriş modülü adını belirtin, örneğin my_job.
   Additional Dependencies	Varsa konektör JAR dosyalarını veya yapılandırma dosyalarını seçin.
   Python Libraries	Varsa üçüncü taraf Python paketlerini (.whl veya .zip) seçin.
   Python Archives	Varsa özel Python sanal ortamını (.zip) seçin.

4. Deploy seçeneğine tıklayın.

Dağıtım parametreleri hakkında daha fazla bilgi için bkz. İş dağıtma.

import logging
import sys

from pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

logging.basicConfig(stream=sys.stdout, level=logging.INFO)

def kafka_to_mysql():
    # Create the execution environment.
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)

    # Create the Kafka source table.
    t_env.execute_sql("""
        CREATE TABLE kafka_source (
            `id` INT,
            `name` STRING,
            `score` INT,
            `event_time` TIMESTAMP(3),
            WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
        ) WITH (
            'connector' = 'kafka',
            'topic' = 'student_topic',
            'properties.bootstrap.servers' = 'your-kafka-broker:9092',
            'properties.group.id' = 'my-group',
            'scan.startup.mode' = 'latest-offset',
            'format' = 'json'
        )
    """)

    # Create the MySQL sink table.
    t_env.execute_sql("""
        CREATE TABLE mysql_sink (
            `id` INT,
            `name` STRING,
            `score` INT,
            PRIMARY KEY (id) NOT ENFORCED
        ) WITH (
            'connector' = 'jdbc',
            'url' = 'jdbc:mysql://your-mysql-host:3306/my_database',
            'table-name' = 'student',
            'username' = 'your_username',
            'password' = 'your_password'
        )
    """)

    # Filter records with scores greater than 60 and write them to MySQL.
    t_env.execute_sql("""
        INSERT INTO mysql_sink
        SELECT id, name, score
        FROM kafka_source
        WHERE score >= 60
    """)

if __name__ == '__main__':
    kafka_to_mysql()