Python - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

Realtime Compute for Apache Flink mendukung user-defined function berbasis Python dalam pekerjaan Flink SQL. Topik ini menjelaskan jenis-jenis user-defined function berbasis Python, cara menggunakan dependensi Python, serta metode untuk tuning kinerja.

Jenis user-defined function

Category	Description
user-defined scalar function (UDSF)	UDSF memetakan nol, satu, atau beberapa nilai skalar menjadi nilai skalar baru. Fungsi ini memproses satu baris input untuk menghasilkan satu nilai output, sehingga menciptakan pemetaan satu-ke-satu. Untuk informasi selengkapnya, lihat User-defined scalar functions (UDSFs).
user-defined aggregate function (UDAF)	UDAF mengagregasi beberapa catatan menjadi satu catatan, sehingga menciptakan pemetaan banyak-ke-satu. Untuk informasi selengkapnya, lihat User-defined aggregate functions (UDAFs).
user-defined table function (UDTF)	UDTF menerima nol, satu, atau beberapa nilai skalar sebagai parameter input. Berbeda dengan fungsi skalar, fungsi tabel dapat mengembalikan sejumlah baris sebagai output, bukan hanya satu nilai saja. Baris yang dikembalikan dapat terdiri dari satu atau beberapa kolom. Satu panggilan fungsi dapat menghasilkan beberapa baris atau kolom. Untuk informasi selengkapnya, lihat User-defined table functions (UDTFs).

Menggunakan dependensi Python

Kluster Realtime Compute for Apache Flink telah dilengkapi paket Python umum yang dipra-instal, seperti Pandas, NumPy, dan PyArrow. Untuk daftar lengkap paket Python pihak ketiga yang telah dipra-instal, lihat halaman Develop Python jobs. Anda harus mengimpor paket-paket tersebut dalam fungsi Anda sebelum menggunakannya. Contoh berikut menunjukkan cara mengimpor sebuah paket.

@udf(result_type=DataTypes.FLOAT())
def percentile(values: List[float], percentile: float):
    import numpy as np
    return np.percentile(values, percentile)

Untuk menggunakan paket Python pihak ketiga yang tidak dipra-instal, unggah paket tersebut sebagai file dependensi saat mendaftarkan Python UDF. Untuk informasi selengkapnya, lihat Manage user-defined functions (UDFs) dan Use Python dependencies.

Debugging kode

Anda dapat menggunakan modul logging untuk mengeluarkan informasi log dari user-defined function berbasis Python Anda. Hal ini membantu Anda menemukan dan mendiagnosis masalah. Contoh berikut menunjukkan cara menggunakan logging.

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

Setelah log dihasilkan, Anda dapat melihatnya dalam file log TaskManager. Untuk informasi selengkapnya, lihat View run logs.

Tuning kinerja

Preloading resources

Pra-muat sumber daya selama inisialisasi fungsi untuk menghindari pemuatan ulang pada setiap pemanggilan metode eval. Misalnya, Anda dapat memuat model pembelajaran mendalam yang besar hanya sekali, lalu menjalankan prediksi batch pada model tersebut berkali-kali.

from pyflink.table import DataTypes
from pyflink.table.udf import ScalarFunction, udf

class Predict(ScalarFunction):
    def open(self, function_context):
        import pickle

        with open("resources.zip/resources/model.pkl", "rb") as f:
            self.model = pickle.load(f)

    def eval(self, x):
        return self.model.predict(x)

predict = udf(Predict(), result_type=DataTypes.DOUBLE(), func_type="pandas")

Catatan

Untuk informasi tentang cara mengunggah file data Python, lihat Menggunakan dependensi Python.

Asynchronous user-defined functions

Untuk skenario intensif I/O, seperti mengakses database eksternal atau melakukan panggilan layanan HTTP, gunakan user-defined function asinkron. Satu instans fungsi dapat menangani beberapa permintaan secara konkuren, mendistribusikan waktu tunggu di beberapa panggilan, dan meningkatkan throughput pekerjaan secara signifikan. Fitur ini hanya didukung di VVR 11.7 dan versi yang lebih baru serta hanya berlaku untuk user-defined skalar function (UDSF). Untuk detailnya, lihat User-defined function asinkron.

Menggunakan library Pandas

Selain user-defined function Python standar, Realtime Compute for Apache Flink juga mendukung Pandas user-defined function. Untuk fungsi-fungsi ini, tipe data input berupa struktur data Pandas, seperti pandas.Series dan pandas.DataFrame. Anda dapat memanfaatkan library Python berkinerja tinggi seperti Pandas dan NumPy dalam fungsi-fungsi ini untuk mengembangkan user-defined function Python yang efisien. Untuk informasi selengkapnya, lihat Vectorized User-defined Functions.

Parameter

Kinerja user-defined function berbasis Python sangat bergantung pada implementasinya. Jika Anda mengalami masalah kinerja, optimalkan implementasi fungsi tersebut. Kinerja juga dipengaruhi oleh parameter-parameter berikut.

Parameter	Description
python.fn-execution.bundle.size	Python UDF dieksekusi secara asinkron. Operator Java menyimpan cache data sebelum mengirimkannya ke proses Python untuk dieksekusi. Ketika cache mencapai ambang batas tertentu, data dikirim ke proses Python. Parameter `python.fn-execution.bundle.size` mengontrol jumlah maksimum catatan data yang dapat disimpan dalam cache. Nilai default-nya adalah 100.000 catatan.
python.fn-execution.bundle.time	Parameter ini mengontrol waktu caching maksimum untuk data. Data yang di-cache dikirim untuk diproses ketika jumlah catatan mencapai ambang batas `python.fn-execution.bundle.size` atau waktu caching mencapai ambang batas `python.fn-execution.bundle.time`. Nilai default-nya adalah 1.000 milidetik.
python.fn-execution.arrow.batch.size	Untuk Pandas UDF, parameter ini menentukan jumlah maksimum catatan yang dapat dimuat dalam satu batch Arrow. Nilai default-nya adalah 10.000. Catatan Nilai parameter `python.fn-execution.arrow.batch.size` tidak boleh lebih besar daripada nilai parameter `python.fn-execution.bundle.size`.

Catatan

Mengatur parameter-parameter ini ke nilai yang terlalu besar justru dapat merugikan. Hal ini dapat menyebabkan terlalu banyak data dibuffer selama Checkpoint, sehingga menyebabkan Checkpoint berlangsung lama atau gagal. Untuk informasi lebih lanjut tentang parameter-parameter ini, lihat Configuration.

Topik terkait

Untuk informasi tentang cara mendaftarkan, memperbarui, dan menghapus user-defined function, lihat Manage user-defined functions (UDFs).
Untuk demo pengembangan dan penggunaan user-defined function berbasis Python, lihat User-defined aggregate functions (UDAFs), User-defined scalar functions (UDSFs), dan User-defined table functions (UDTFs).
Untuk mempelajari cara menggunakan lingkungan virtual Python kustom, paket Python pihak ketiga, paket JAR, dan file data dalam pekerjaan Python Flink, lihat Use Python dependencies.
Untuk demo pengembangan dan penggunaan user-defined function berbasis Java, lihat User-defined aggregate functions (UDAFs), User-defined scalar functions (UDSFs), dan User-defined table functions (UDTFs).
Untuk mempelajari cara debugging dan tuning user-defined function berbasis Java, lihat Overview.
Implement data sorting and aggregation using a UDAF.