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

Batasan

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

Metode pengembangan

Anda dapat mengembangkan UDSF Python dengan salah satu cara berikut:

• Kemas kode Python Anda, unggah paket tersebut, lalu daftarkan sebagai fungsi di platform.

• Deklarasikan logika fungsi secara inline dalam pernyataan SQL.

Jika logika UDSF Anda sederhana, pertimbangkan untuk mengembangkannya sebagai fungsi inline.

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, lalu buka direktori python_demo-master yang telah diekstrak.

3. Buka udfs.py pada path \python_demo-master\udx dan 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 sub_string mengekstraksi karakter dari posisi begin hingga posisi end pada string input.

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

   zip -r python_demo.zip udx

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

Mendaftarkan UDSF

Setelah membuat paket, daftarkan UDSF di konsol Realtime Compute for Apache Flink. Untuk langkah-langkah pendaftaran, 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 penerapan tersebut, lalu klik Start di kolom Actions. Setelah penerapan dimulai, karakter pada posisi 2–4 dari bidang a di ASI_UDSF_Source akan ditulis ke ASI_UDSF_Sink.

Fungsi skalar inline Python

Fungsi inline menyematkan implementasinya langsung dalam pernyataan CREATE FUNCTION, sehingga fungsi tersebut didefinisikan dan didaftarkan dalam satu pernyataan SQL yang sama. Contoh berikut mendefinisikan fungsi inline yang menyamarkan alamat email. Deklarasikan seluruh logika fungsi di antara pembatas $$.

CREATE TEMPORARY FUNCTION mask_email(email STRING)
RETURNS STRING AS $$
if email is None:
    return None

name, separator, domain = email.partition("@")
if not separator:
    return email

return name[:1] + "***@" + domain
$$ LANGUAGE PYTHON;

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 konkuren, membagi waktu tunggu di antara permintaan, dan meningkatkan throughput pekerjaan.

import asyncio

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


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


# Method 2: Subclass the asynchronous function class
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()
)