Realtime Compute for Apache Flink：Python作業開發

Table API/SQL與DataStream API的選擇

PyFlink支援Table API/SQL和DataStream API兩種開發方式，推薦優先使用Table API/SQL。原因如下：

• 效能更優：Table API/SQL的執行計畫經過最佳化後完全在JVM內執行。而DataStream API需要在JVM和Python進程之間逐條資料進行序列化/還原序列化，效能開銷較大。

• 功能更完善：Table API/SQL對連接器、資料格式、視窗函數等支援更完整，且與SQL作業共用同一套連接器生態。

• 社區推薦：Apache Flink社區在PyFlink的發展方向上以Table API/SQL為主。

只有在SQL無法表達的複雜自訂邏輯情境下，才建議使用DataStream API。

開發參考

您可以參見以下文檔在本地完成Flink業務代碼開發，開發完成後您需要將其上傳到Flink開發控制台，並部署上線作業。

• Apache Flink V1.20業務代碼開發，請參見Flink Python API開發指南。

• Apache Flink編碼過程中遇到的問題及解決方案，請參見常見問題。

專案結構

推薦的Python作業專案結構如下：

my-flink-python-project/
├── my_job.py                # 主作業檔案
├── udfs.py                  # 自訂函數（可選）
├── requirements.txt         # 第三方Python依賴（可選）
└── config.properties        # 設定檔（可選）

依賴管理

Python作業中使用自訂的Python虛擬環境、第三方Python包、JAR包和資料檔案的方法，請參見使用Python依賴。

自訂函數（UDF）

以下是一個Python UDSF的開發樣本，用於對字串進行脫敏處理：

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())def mask_phone(phone: str):"""手機號脫敏：保留前3位和後4位，中間用****替代"""if phone is None or len(phone) != 11:return phone
    return phone[:3] + '****' + phone[7:]

在SQL作業中使用該UDF：

CREATE TEMPORARY FUNCTION mask_phone AS 'udfs.mask_phone' LANGUAGE PYTHON;INSERT INTO sink_table
SELECT name, mask_phone(phone) AS masked_phone
FROM source_table;

UDF的註冊、更新和刪除方法，請參見管理自訂函數（UDF）。

連接器使用

Flink所支援的連接器列表，請參見支援的連接器。連接器使用方法如下：

1. 登入Realtime Compute控制台。

2. 單擊目標工作空間操作列下的控制台。

3. 在左側導覽列，單擊檔案管理。

4. 單擊上傳資源，選擇您要上傳的目標連接器的Python包。

   您可以上傳自己開發的連接器，也可以上傳Flink提供的連接器。Flink提供的連接器官方Python包的下載地址，請參見Connector列表。

5. 在營運中心 > 作業營運頁面，單擊部署作業 > Python 作業，附加依賴檔案項選擇目標連接器的Python包，配置其他參數並部署作業。

6. 單擊部署的作業名稱，在部署詳情頁簽運行參數配置地區，單擊編輯，在其他配置中，添加Python連接器包位置資訊。

   如果您的作業需要依賴多個連接器Python包，例如依賴的2個包的名字分別為connector-1.jar和connector-2.jar，則配置資訊如下。

   pipeline.classpaths: 'file:///flink/usrlib/connector-1.jar;file:///flink/usrlib/connector-2.jar'

7. 如果您需要使用內建連接器、資料格式和Catalog（僅VVR 11.2及以上版本），可在作業的運行參數配置地區其他配置項中添加配置，例如：

   ## 多個連接器使用
   pipeline.used-builtin-connectors: kafka;sls
   ## 傳輸資料的多種資料格式
   pipeline.used-builtin-formats: avro;parquet
   ## 使用了多個已經建立的Catalog
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

具體的連接器使用方式詳情請參見完整範例程式碼。

作業調試

import logging

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

本地調試

由於Realtime ComputeFlink版預設不具備訪問公網的能力，您的代碼可能無法在本地直接連接線上資料來源進行測試。建議按照以下方式進行本地調試：

• 單元測試：對自訂函數（UDF）進行獨立的單元測試，確保函數邏輯正確。

• 本地運行：使用本機資料源（如檔案或記憶體資料）類比輸入，在本地運行作業驗證處理邏輯。樣本如下：

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()# 使用本機資料源進行測試
  ds = env.from_collection([('Alice', 1), ('Bob', 2), ('Alice', 3)])
  ds.key_by(lambda x: x[0]).sum(1).print()
  env.execute("local_test")

• 遠端偵錯：如需連接線上資料來源調試，請參見本地運行和調試包含連接器的作業。

作業部署

Python作業開發完成後，需要上傳到Realtime Compute控制台進行部署。操作步驟如下：

1. 登入Realtime Compute控制台，進入目標工作空間。

2. 在左側導覽列單擊檔案管理，上傳Python作業檔案（.py或.zip）。如有第三方依賴或設定檔，也需一併上傳。

3. 在營運中心 > 作業營運頁面，單擊部署作業 > Python作業，填寫部署資訊。

   參數	說明
   Python檔案地址	選擇已上傳的Python作業檔案。
   Entry Module	如果作業檔案為.py檔案，無需填寫；如果為.zip檔案，需填寫入口模組名稱，例如my_job。
   附加依賴檔案	如有連接器JAR包或設定檔，在此選擇。
   Python Libraries	如有第三方Python包（.whl或.zip），在此選擇。
   Python Archives	如有自訂Python虛擬環境（.zip），在此選擇。

4. 單擊部署。

更多部署參數詳情請參見部署作業。