Buat, daftarkan, dan gunakan Python UDTF - Realtime Compute for Apache Flink

Topik ini menjelaskan cara mengembangkan, mendaftarkan, dan menggunakan fungsi bernilai tabel yang didefinisikan pengguna (user-defined table-valued functions/UDTFs) berbasis Python.

Definisi

Fungsi bernilai tabel yang didefinisikan pengguna (user-defined table-valued function/UDTF) menerima nol, satu, atau lebih nilai skalar sebagai parameter input—yang dapat memiliki panjang variabel. Berbeda dengan fungsi skalar, UDTF dapat mengembalikan sejumlah baris apa pun, bukan hanya satu nilai. Setiap baris yang dikembalikan dapat berisi satu atau lebih kolom, dan setiap pemanggilan UDTF dapat menghasilkan beberapa baris atau kolom.

Batasan

Karena lingkungan penerapan dan jaringan Realtime Compute for Apache Flink, perhatikan batasan berikut saat mengembangkan fungsi yang didefinisikan pengguna berbasis Python:

Hanya Flink open source versi 1.12 ke atas yang didukung.
Python telah dipra-instal di ruang kerja Flink. Kembangkan kode Anda menggunakan versi Python yang telah dipra-instal tersebut.

Catatan

Mesin Realtime Compute VVR versi sebelum 8.0.11 telah mempra-instal Python 3.7.9, sedangkan VVR 8.0.11 ke atas telah mempra-instal Python 3.9.21. Jika Anda melakukan peningkatan dari versi sebelumnya ke VVR 8.0.11 atau yang lebih baru, Anda harus menguji ulang, menerapkan, dan menjalankan pekerjaan PyFlink yang dikembangkan pada versi sebelumnya.

Waktu proses Flink hanya mendukung JDK 8 dan JDK 11. Jika pekerjaan Python bergantung pada paket JAR pihak ketiga, pastikan paket tersebut kompatibel.
Hanya Scala open source versi 2.11 yang didukung. Jika pekerjaan Python bergantung pada paket JAR pihak ketiga, gunakan paket JAR yang sesuai dengan Scala V2.11.
Fungsi inline hanya didukung pada VVR 11.9.0 ke atas.

Kembangkan UDTF

Anda dapat mengembangkan fungsi Python dengan salah satu cara berikut:

Paket dan unggah kode Python, lalu daftarkan kode tersebut di platform.
Deklarasikan logika kode sebagai fungsi inline dalam skrip SQL.

Jika logika UDTF sederhana, Anda dapat mengembangkannya sebagai fungsi inline.

Kembangkan UDTF menggunakan paket kode

Kembangkan UDTF

Catatan

Flink menyediakan proyek contoh ekstensi yang didefinisikan pengguna (UDXs) berbasis Python untuk membantu Anda mengembangkan UDX. Contoh tersebut mencakup implementasi fungsi skalar yang didefinisikan pengguna, fungsi agregat yang didefinisikan pengguna (UDAF), dan UDTF. Contoh ini menggunakan sistem operasi Windows untuk menjelaskan cara mengembangkan UDTF.

Penting

Nama UDF yang terdaftar hanya boleh berisi huruf kecil, angka, dan tanda hubung (-). Karakter garis bawah (_) dan karakter khusus lainnya tidak didukung.
Nama fungsi tidak boleh sama dengan nama fungsi bawaan. Lingkungan terkelola Realtime Compute for Apache Flink telah mempra-instal beberapa fungsi bawaan, seperti split. Jika fungsi kustom memiliki nama yang sama dengan fungsi bawaan, pendaftaran akan gagal atau validasi SQL akan melaporkan error. Gunakan awalan kustom untuk nama fungsi, misalnya my-split.

Unduh dan ekstrak paket contoh python\_demo-master ke komputer Anda.
Di PyCharm, pilih File > Open untuk membuka folder python\_demo-master yang telah diekstrak.
Klik ganda \python\_demo-master\udx\udtfs.py dan modifikasi isi file udtfs.py sesuai kebutuhan bisnis Anda.

Pada contoh ini, my\_split memisahkan string setiap baris menjadi beberapa kolom berdasarkan tanda pipa vertikal (|).


    from pyflink.table import DataTypes
    from pyflink.table.udf import udtf
    
    @udtf(result_types=[DataTypes.STRING(), DataTypes.STRING()])
    def my_split(s: str):
        splits = s.split("|")
        yield splits[0], splits[1]

Di direktori yang berisi folder udx dalam paket yang diunduh (\python\_demo), jalankan perintah berikut untuk membuat paket file.


    zip -r python_demo.zip udx

Paket python\_demo.zip dihasilkan di direktori \python\_demo\, yang menandakan bahwa UDTF telah dikembangkan.

Daftarkan UDTF

Untuk informasi selengkapnya tentang cara mendaftarkan UDTF, lihat Manage UDFs.

Gunakan UDTF

Setelah UDTF terdaftar, Anda dapat menggunakannya. Lakukan langkah-langkah berikut:

Kembangkan pekerjaan Flink SQL. Untuk informasi selengkapnya, lihat Job development overview.

Nilai bidang message di setiap baris tabel ASI\_UDTF\_Source digabungkan dengan string aa menggunakan tanda pipa vertikal (|), lalu dipisahkan menjadi beberapa kolom berdasarkan tanda pipa vertikal (|). Kode berikut memberikan contohnya:


    CREATE TEMPORARY TABLE ASI_UDTF_Source (
      `message`  VARCHAR
    ) WITH (
      'connector'='datagen'
    );
    
    CREATE TEMPORARY TABLE ASI_UDTF_Sink (
      name  VARCHAR,
      place  VARCHAR
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO ASI_UDTF_Sink
    SELECT name,place
    FROM ASI_UDTF_Source,lateral table(my_split(concat_ws('|', `message`, 'aa'))) as T(name,place);

Di halaman O&M > Deployments, klik Start di kolom Actions pekerjaan yang dituju.

Setelah pekerjaan dimulai, tabel ASI\_UDTF\_Sink berisi dua kolom. Data dihasilkan dengan menggabungkan bidang message dan aa dari setiap baris di tabel ASI\_UDTF\_Source menggunakan tanda pipa vertikal (|), lalu memisahkan hasilnya berdasarkan tanda pipa vertikal (|).

Fungsi tabel inline Python

Fungsi inline menyematkan implementasi fungsi secara langsung dalam pernyataan CREATE FUNCTION. Definisi fungsi dan deklarasi pendaftaran diselesaikan dalam satu pernyataan SQL yang sama. Contoh berikut mendefinisikan fungsi inline yang menyamarkan alamat email. Anda harus mendeklarasikan logika kode lengkap di antara $$.

CREATE TEMPORARY FUNCTION parse_order_items(items_json STRING)
RETURNS TABLE (
    sku STRING,
    quantity INT
)
HANDLER 'ParseOrderItems'
AS $$
import json

class ParseOrderItems:
    def eval(self, items_json):
        if not items_json:
            return

        try:
            items = json.loads(items_json)
        except (TypeError, ValueError):
            return

        if not isinstance(items, list):
            return

        for item in items:
            if not isinstance(item, dict):
                continue

            sku = str(item.get("sku", "")).strip().upper()

            try:
                quantity = int(item.get("quantity"))
            except (TypeError, ValueError):
                continue

            if sku and quantity > 0:
                yield sku, quantity
$$
LANGUAGE PYTHON;

Catatan

Python sensitif terhadap indentasi. Tulis badan fungsi tingkat pertama di awal baris, dan pastikan indentasi relatif di dalam blok kode sudah benar.
Fungsi inline tidak mendukung eksekusi vektorisasi dan tidak dapat dideklarasikan sebagai fungsi nondeterministik.
Hanya fungsi TEMPORARY yang dapat dibuat.