Realtime Compute for Apache Flink:Mengembangkan pekerjaan PyFlink

Pilih antara Table API/SQL dan DataStream API

PyFlink mendukung Table API/SQL dan DataStream API. Kami merekomendasikan menggunakan Table API/SQL terlebih dahulu. Berikut alasannya:

• Kinerja lebih baik: Rencana eksekusi Table API/SQL berjalan sepenuhnya di dalam JVM setelah dioptimalkan. DataStream API memerlukan serialisasi dan deserialisasi per-record antara proses JVM dan Python, yang menambahkan overhead signifikan.

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

• Rekomendasi komunitas: Komunitas Apache Flink memprioritaskan Table API/SQL untuk pengembangan PyFlink.

Gunakan DataStream API hanya jika SQL tidak dapat mengekspresikan logika kustom yang kompleks.

Referensi pengembangan

Gunakan sumber daya berikut untuk mengembangkan logika bisnis Flink secara lokal. Setelah selesai, unggah kode ke Konsol Realtime Compute for Apache Flink dan terapkan pekerjaan tersebut.

• Untuk cara mengembangkan logika bisnis Apache Flink 1.20, lihat Panduan Pengembangan API Python Flink.

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

Struktur proyek

Struktur proyek pekerjaan Python yang direkomendasikan adalah sebagai berikut:

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 cara menggunakan lingkungan virtual Python kustom, paket Python pihak ketiga, JAR, dan file data dalam pekerjaan PyFlink, lihat Menggunakan dependensi Python.

Fungsi yang ditentukan pengguna (UDF)

Contoh berikut menunjukkan cara mengembangkan UDSF Python untuk 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 tiga digit pertama dan empat digit terakhir; ganti digit tengah dengan ****."""
    if phone is None or len(phone) != 11:
        return phone
    return phone[:3] + '****' + phone[7:]

Untuk menggunakan UDF ini dalam pekerjaan SQL:

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 cara mendaftarkan, memperbarui, dan menghapus UDF, lihat Mengelola fungsi yang ditentukan pengguna (UDF).

Menggunakan konektor

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

1. Masuk ke Konsol Realtime Compute for Apache Flink.

2. Klik Console di kolom Actions ruang kerja target.

3. Di panel navigasi kiri, klik File Management.

4. Klik Upload Artifact, lalu pilih paket Python untuk konektor target.

   Anda dapat mengunggah konektor kustom atau konektor bawaan dari Realtime Compute for Apache Flink. Untuk tautan unduhan paket Python resmi untuk konektor Flink, lihat Daftar konektor.

5. Di halaman Operation Center > Job O&M, klik Deploy Job > Python Job, pilih paket Python untuk konektor target di kolom Additional Dependencies, konfigurasikan parameter lainnya, lalu terapkan pekerjaan tersebut.

6. Klik nama pekerjaan yang telah diterapkan. Di tab Deployment Details, pada bagian Runtime Parameter Settings, klik Edit, lalu di Other Configuration, tambahkan path ke paket konektor Python.

   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, atau katalog (hanya tersedia di VVR 11.2 dan yang lebih baru), tambahkan konfigurasi di kolom Other Configuration di bawah Parameters. Contohnya:

   ## Gunakan beberapa konektor bawaan
   pipeline.used-builtin-connectors: kafka;sls
   ## Gunakan beberapa format data
   pipeline.used-builtin-formats: avro;parquet
   ## Gunakan beberapa katalog yang ada
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

Untuk contoh penggunaan konektor secara detail, lihat Kode contoh lengkap.

Men-debug Pekerjaan

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 Internet. Kode Anda mungkin tidak dapat terhubung langsung ke sumber data online untuk pengujian. Kami merekomendasikan pendekatan berikut untuk debugging lokal:

• Pengujian unit: Jalankan pengujian unit independen pada UDF Anda untuk memverifikasi kebenaran logika.

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

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()
  # Uji menggunakan sumber data lokal
  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 dengan konektor secara lokal.

Menyebarkan pekerjaan

Setelah mengembangkan pekerjaan PyFlink, unggah ke Konsol Realtime Compute for Apache Flink untuk diterapkan. Ikuti langkah-langkah berikut:

1. Masuk ke Konsol Realtime Compute for Apache Flink dan buka ruang kerja target.

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

3. Di halaman Operation Center > Job O&M, klik Deploy Job > Python Job, lalu isi informasi penyebaran.

   Parameter	Deskripsi
   Python file path	Pilih file pekerjaan Python yang telah diunggah.
   Entry Module	Biarkan kosong untuk file .py. Untuk file .zip, masukkan nama modul entri—misalnya, my_job.
   Additional Dependencies	Pilih JAR konektor atau file konfigurasi, jika ada.
   Python Libraries	Pilih paket Python pihak ketiga (.whl atau .zip), jika ada.
   Python Archives	Pilih lingkungan virtual Python kustom (.zip), jika ada.

4. Klik Deploy.

Untuk detail lebih lanjut tentang parameter penyebaran, lihat Menyebarkan pekerjaan.