Python自訂純量涵式UDSF的開發、註冊和使用流程 - Realtime Compute for Apache Flink

本文為您介紹Python自訂純量涵式（UDSF）的開發、註冊和使用流程。

定義

自訂純量涵式（UDSF）將0個、1個或多個標量值對應到一個新的標量值。輸入與輸出是一對一的關係，即讀入一行資料，寫出一條輸出值。

使用限制

由於Realtime ComputeFlink版受部署環境和網路環境等因素的影響，開發Python自訂函數時，需要注意以下限制：

僅支援開源Flink V1.12及以上版本。
Flink工作空間已預裝了Python，因此需要您在對應Python版本上開發代碼。
說明
Realtime Compute引擎VVR 8.0.11以下版本預裝Python 3.7.9版本，Realtime Compute引擎VVR 8.0.11及以上版本預裝Python 3.9.21版本。如需將低版本升級至Realtime Compute引擎VVR 8.0.11及以上版本，必須對先前版本的PyFlink作業進行重新測試、部署和運行。
Flink運行環境僅支援JDK 8和JDK 11，如果Python作業中依賴第三方JAR包，請確保JAR包相容。
僅支援開源Scala V2.11版本，如果Python作業中依賴第三方JAR包，請確保使用Scala V2.11對應的JAR包依賴。

UDSF開發

說明

Flink為您提供了Python自訂函數樣本，便於您快速開發自訂函數。Flink Python自訂函數樣本中包含了Python UDSF、Python UDAF和Python UDTF的實現。本文以Windows作業系統為例，為您介紹如何進行UDSF開發。

下載並解壓python_demo-master樣本到本地。
說明
python_demo-master屬於第三方搭建的網站，訪問時可能會存在無法開啟或訪問延遲的問題。
在PyCharm中，單擊file > open，開啟剛才解壓縮完成的python_demo-master。

雙擊開啟\python_demo-master\udx\udfs.py後，根據您的業務，修改udfs.py。

該樣本中，sub_string定義了擷取每條資料中從begin~end位的字元的代碼。

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]

在下載檔案中udx所在的目錄（即\python_demo-master目錄）下執行如下命令打包檔案。
```
zip -r python_demo.zip udx
```
\python_demo-master\目錄下會出現python_demo.zip的ZIP包，即代表完成了Python UDSF的開發工作。

UDSF註冊

UDSF註冊過程，請參見管理自訂函數（UDF）。

UDSF使用

在完成註冊UDSF後，您就可以使用UDSF，詳細的操作步驟如下。

Flink SQL作業開發。詳情請參見作業開發地圖。

擷取ASI_UDSF_Source表中a欄位中每行字串中第2~4位的字元，程式碼範例如下。

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;

在營運中心 > 作業營運頁面，單擊目標作業名稱操作列的啟動。
啟動成功後，ASI_UDSF_Sink表每行會被插入ASI_UDSF_Source表中a欄位每行字串的第2~4位字元。

非同步自訂函數

如果自訂函數中需要進行外部資料庫訪問、HTTP 服務等 I/O 密集型操作，建議使用非同步自訂函數。單個非同步自訂函數可以並發處理多個 I/O 請求，使得等待時間分攤到多個請求上，提升作業吞吐。

使用限制

僅 VVR 11.7 及以上版本支援，需安裝VVR pyflink >=11.7。詳情請參見ververica-flink。
pip3 install "ververica-flink>=11.7"
僅支援非同步自訂純量涵式（UDSF）。
僅支援 Python 進程模式，即 python.execution-mode=process。
暫不支援 Pandas 非同步自訂函數。

使用方式

非同步自訂函數可以通過 Python 非同步函數或繼承非同步函數類實現，範例程式碼如下。

import asyncio

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


# 方法一：使用 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}"


# 方法二：繼承非同步函數類
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()
)

非同步自訂函數的註冊與使用方式與同步函數相同。

配置參數

您可以通過修改以下參數改變非同步自訂函數的運行行為。

參數	預設值	說明
table.exec.async-scalar.max-concurrent-operations	10	單個運算元執行個體允許同時觸發的最大非同步呼叫數。預設值為 10。
table.exec.async-scalar.timeout	3 min	單次非同步呼叫的逾時時間。
table.exec.async-scalar.retry-strategy	FIXED_DELAY	非同步呼叫失敗後的重試策略，支援： FIXED_DELAY：等待固定時間後重試。 NO_RETRY：不進行重試。
table.exec.async-scalar.retry-delay	100 ms	固定延遲重試的等待時間。說明僅當 table.exec.async-scalar.retry-strategy 為 FIXED_DELAY 時生效。
table.exec.async-scalar.max-attempts	3	非同步呼叫失敗前的最大嘗試次數。說明僅當 table.exec.async-scalar.retry-strategy 為 FIXED_DELAY 時生效。