Realtime Compute for Apache Flink:Fungsi skalar yang didefinisikan pengguna Python (UDSFs)

Batasan

Batasan berikut berlaku saat Anda mengembangkan fungsi yang didefinisikan pengguna (UDF) Python di Realtime Compute for Apache Flink:

Membuat UDSF

Langkah-langkah berikut menggunakan Windows sebagai contoh lingkungan. Flink menyediakan repositori contoh yang mencakup implementasi untuk UDSF, fungsi agregat yang didefinisikan pengguna (UDAF), dan fungsi bernilai tabel yang didefinisikan pengguna (UDTF).

1. Unduh dan ekstrak python_demo-master ke mesin lokal Anda.

   Ini adalah repositori GitHub pihak ketiga. Akses mungkin lambat atau tidak stabil.

2. Di PyCharm, pilih File > Open dan buka direktori python_demo-master yang telah diekstrak.

3. Buka file udfs.py di path \python_demo-master\udx, lalu definisikan UDSF Anda.

   from pyflink.table import DataTypes
   from pyflink.table.udf import udf
   
   @udf(result_type=DataTypes.STRING())
   def sub_string(s: str, begin: int, end: int):
       return s[begin:end]

   Contoh fungsi sub_string mengekstraksi karakter dari posisi begin hingga posisi end dalam string input.

4. Dari direktori \python_demo-master, jalankan perintah berikut untuk membuat paket direktori udx:

   zip -r python_demo.zip udx

   Saat file python_demo.zip muncul di \python_demo-master\, paket tersebut siap digunakan.

Mendaftarkan UDSF

Setelah membuat paket, daftarkan UDSF di konsol Realtime Compute for Apache Flink. Untuk langkah-langkah pendaftarannya, lihat Mengelola fungsi yang didefinisikan pengguna (UDF).

Menggunakan UDSF

Setelah mendaftarkan UDSF, gunakan fungsi tersebut dalam pekerjaan Flink SQL.

1. Buat draft menggunakan Flink SQL. Untuk detailnya, lihat Ikhtisar pengembangan pekerjaan. Contoh berikut memanggil ASI_UDSF (nama terdaftar UDSF Anda) untuk mengekstraksi karakter dari posisi 2 hingga 4 pada bidang a di tabel sumber:

   CREATE TEMPORARY TABLE ASI_UDSF_Source (
     a VARCHAR,
     b INT,
     c INT
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDSF_Sink (
     a VARCHAR
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDSF_Sink
   SELECT ASI_UDSF(a, 2, 4)
   FROM ASI_UDSF_Source;

2. Di panel navigasi sebelah kiri Konsol pengembangan, pilih O&M > Deployments. Temukan deployment tersebut, lalu klik Start di kolom Actions. Setelah deployment dimulai, karakter pada posisi 2–4 dari bidang a di ASI_UDSF_Source akan ditulis ke ASI_UDSF_Sink.

Fungsi yang didefinisikan pengguna asinkron

Untuk UDF yang melakukan operasi intensif I/O seperti akses database eksternal atau permintaan HTTP, gunakan fungsi yang didefinisikan pengguna asinkron. Satu UDF asinkron dapat menangani beberapa permintaan I/O secara bersamaan, mendistribusikan waktu tunggu di antara permintaan, serta meningkatkan throughput pekerjaan.

import asyncio

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


# Metode 1: Gunakan fungsi async Python
@udf(result_type=DataTypes.STRING())
async def async_api_call(product_id: str) -> str:
    await asyncio.sleep(0.05)
    return f"product_{product_id}"


# Metode 2: Subclass dari kelas fungsi asinkron
class AsyncUserLookup(AsyncScalarFunction):
    def open(self, function_context):
        self.cache = {}

    async def eval(self, user_id: str) -> str:
        if user_id in self.cache:
            return self.cache[user_id]

        await asyncio.sleep(0.05)
        result = f"user_{user_id}"
        self.cache[user_id] = result
        return result

    def close(self):
        self.cache.clear()


async_user_lookup = udf(
    AsyncUserLookup(),
    input_types=[DataTypes.STRING()],
    result_type=DataTypes.STRING()
)