Realtime Compute for Apache Flink：Pengembangan Pekerjaan Python

Memilih antara Table API/SQL dan DataStream API

PyFlink mendukung Table API/SQL maupun DataStream API. Kami merekomendasikan menggunakan Table API/SQL terlebih dahulu karena alasan berikut:

• Kinerja lebih baik: Rencana eksekusi Table API/SQL dioptimalkan dan dijalankan sepenuhnya dalam JVM. DataStream API memerlukan serialisasi dan deserialisasi setiap record antara proses JVM dan Python, yang menimbulkan overhead signifikan.

• Fitur lebih lengkap: Table API/SQL menawarkan dukungan lebih lengkap untuk konektor, format data, dan fungsi jendela, serta berbagi ekosistem konektor yang sama dengan pekerjaan SQL.

• Arah komunitas: Komunitas Apache Flink memfokuskan pengembangan PyFlink terutama pada Table API/SQL.

Gunakan DataStream API hanya untuk logika kustom kompleks yang tidak dapat diekspresikan dalam SQL.

Referensi pengembangan

Gunakan dokumen berikut untuk mengembangkan kode bisnis Flink secara lokal. Setelah pengembangan selesai, unggah kode Anda ke Konsol pengembangan Flink dan terapkan pekerjaan tersebut.

• Untuk pengembangan kode bisnis Apache Flink 1.20, lihat Panduan Pengembangan Flink Python API.

• Untuk masalah umum dan solusinya selama pengkodean Apache Flink, lihat FAQ.

Struktur proyek

Kami merekomendasikan struktur proyek berikut untuk pekerjaan Python:

my-flink-python-project/
├── my_job.py                # File pekerjaan utama
├── udfs.py                  # Fungsi yang ditentukan pengguna (opsional)
├── requirements.txt         # Dependensi Python pihak ketiga (opsional)
└── config.properties        # File konfigurasi (opsional)

Manajemen dependensi

Untuk petunjuk penggunaan lingkungan virtual Python kustom, paket Python pihak ketiga, JAR, dan file data dalam pekerjaan Python, lihat Menggunakan dependensi Python.

Fungsi yang ditentukan pengguna (UDF)

Contoh berikut menunjukkan cara mengembangkan UDSF Python yang mendesensitisasi string:

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

@udf(result_type=DataTypes.STRING())
def mask_phone(phone: str):
    """Menyamarkan nomor telepon: pertahankan 3 digit pertama dan 4 digit terakhir, ganti bagian tengah dengan ****"""
    if phone is None or len(phone) != 11:
        return phone
    return phone[:3] + '****' + phone[7:]

Gunakan UDF ini dalam pekerjaan SQL sebagai berikut:

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;

Untuk petunjuk mendaftarkan, memperbarui, dan menghapus UDF, lihat Mengelola fungsi yang ditentukan pengguna (UDF).

Menggunakan konektor

Untuk daftar konektor yang didukung, lihat Konektor yang didukung. Untuk menggunakan konektor, ikuti langkah-langkah berikut:

1. Masuk ke Konsol Realtime Compute.

2. Klik Console di kolom Actions ruang kerja target.

3. Di panel navigasi kiri, klik File Management.

4. Klik Upload Artifact dan pilih paket Python untuk konektor target Anda.

   Anda dapat mengunggah konektor kustom atau konektor Flink bawaan. Untuk tautan unduhan paket Python konektor resmi Flink, lihat Daftar konektor.

5. Pada halaman O&M > Job O&M, klik Deploy Job > Python Job. Di kolom Additional Dependencies, pilih paket konektor Anda, konfigurasikan parameter lainnya, lalu terapkan pekerjaan tersebut.

6. Klik nama pekerjaan yang telah diterapkan. Pada tab Deployment Details, di bagian Runtime Parameters, klik Edit. Di kolom Other Configuration, tambahkan path ke paket konektor Python Anda.

   Jika pekerjaan Anda bergantung pada beberapa paket konektor—misalnya, connector-1.jar dan connector-2.jar—konfigurasikan sebagai berikut:

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

7. Untuk menggunakan konektor bawaan, format data, dan katalog (hanya untuk VVR 11.2 atau yang lebih baru), tambahkan konfigurasi berikut di kolom Other Configuration pada bagian Runtime Parameters:

   pipeline.used-builtin-connectors: kafka;sls
   pipeline.used-builtin-formats: avro;parquet

Debugging Tugas

import logging

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

Debugging lokal

Secara default, Realtime Compute for Apache Flink tidak dapat mengakses jaringan publik, sehingga kode Anda mungkin tidak dapat langsung terhubung ke sumber data online untuk pengujian. Gunakan salah satu pendekatan berikut untuk debugging lokal:

• Unit test: Uji fungsi yang ditentukan pengguna (UDF) secara terpisah untuk memverifikasi logika.

• Eksekusi lokal: Simulasikan input menggunakan sumber data lokal seperti file atau data in-memory, lalu jalankan pekerjaan secara lokal untuk memvalidasi logika pemrosesan. Contohnya:

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()
  # Gunakan sumber data lokal untuk pengujian
  ds = env.from_collection([('Alice', 1), ('Bob', 2), ('Alice', 3)])
  ds.key_by(lambda x: x[0]).sum(1).print()
  env.execute("local_test")

• Debugging jarak jauh: Untuk debugging dengan sumber data online, lihat Menjalankan dan mendebug pekerjaan berbasis konektor secara lokal.

Deployment pekerjaan

Setelah mengembangkan pekerjaan Python, unggah ke Konsol Realtime Compute untuk deployment. Ikuti langkah-langkah berikut:

1. Masuk ke Konsol Realtime Compute dan buka ruang kerja target Anda.

2. Di panel navigasi kiri, klik File Management. Unggah file pekerjaan Python Anda (.py atau .zip), beserta dependensi pihak ketiga atau file konfigurasi apa pun.

3. Pada halaman O&M > Job O&M, klik Deploy Job > Python Job dan masukkan informasi deployment.

   Parameter	Deskripsi
   Python file path	Pilih file pekerjaan Python yang telah diunggah.
   Entry Module	Biarkan kosong jika file pekerjaan Anda berupa file .py. Jika berupa file .zip, masukkan nama modul entri, misalnya my_job.
   Additional Dependencies	Pilih JAR konektor atau file konfigurasi apa pun di sini.
   Python Libraries	Pilih paket Python pihak ketiga (.whl atau .zip) di sini.
   Python Archives	Pilih lingkungan virtual Python kustom (.zip) di sini.

4. Klik Deploy.

Untuk detail parameter deployment lebih lanjut, lihat Menerapkan pekerjaan.