Realtime Compute for Apache Flink：Python作業開發

背景資訊

Flink Python作業需要您在本地完成開發工作，Python作業開發完成後，再在Flink開發控制台上部署並啟動才能看到業務效果。整體的操作流程詳情請參見Flink Python作業。

開發環境要求

• Realtime Compute引擎VVR 8.0.11以下版本預裝Python 3.7.9，VVR 8.0.11及以上版本預裝Python 3.9.21，VVR 11.7及以上版本預裝 Python 3.9.21、3.10.19 和 3.11.14。

  說明

  建議本地開發環境的Python版本與目標VVR引擎預裝的Python版本保持一致。

  說明

  VVR 11.7及以上版本預設使用Python 3.9。如果您希望使用預裝的Python 3.10或3.11，可按下面的方式配置作業：

  1. 在營運中心 > 作業營運頁面，單擊目標作業名稱。

  2. 在部署詳情頁簽運行參數配置地區，點擊右側編輯，在其他配置項中添加配置（以 Python 3.11 為例）：

     python.executable: python3.11
     python.client.executable: python3.11

• 已安裝PyFlink，且版本與目標VVR引擎版本一致。例如，您在部署頁面選擇的引擎為 vvr-11.7.0-jdk11-flink-1.20，則需安裝：

  pip install ververica-flink==11.7.0

  說明

  VVR 11.5及以下未提供專門的PyFlink包，需要安裝對應版本的開源PyFlink，例如您在部署頁面選擇的引擎為vvr-8.0.9-flink-1.17，則需安裝apache-flink==1.17.*。

  pip install apache-flink==1.17.2

• 已安裝IDE開發工具，推薦PyCharm或VS Code。

• Python作業需要您線上下完成開發，再在Realtime Compute管理主控台上部署並運行。

使用限制

由於Flink受部署環境、網路環境等因素的影響，開發Python作業需要注意以下限制：

• 僅支援開源Flink V1.13及以上版本。

• Flink工作空間已預裝了Python環境，且Python環境中已預裝了Pandas、NumPy、PyArrow等常用的Python庫。詳見本文末尾預裝軟體包列表。

• Flink運行環境僅支援JDK 8和JDK 11，如果Python作業中依賴第三方JAR包，請確保JAR包相容。

• VVR 4.x僅支援開源Scala V2.11版本，VVR 6.x 及以上版本僅支援開源Scala V2.12版本。如果Python作業中依賴第三方JAR包，請確保使用Scala版本對應的JAR包依賴。

作業開發

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

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:]

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;

import logging

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

完整範例程式碼

本樣本展示一個從Kafka讀取資料、進行簡單處理後寫入MySQL的Python流作業，僅供參考。

import logging
import sys

from pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

logging.basicConfig(stream=sys.stdout, level=logging.INFO)


def kafka_to_mysql():
    # 建立執行環境
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)

    # 建立 Kafka 源表
    t_env.execute_sql("""
        CREATE TABLE kafka_source (
            `id` INT,
            `name` STRING,
            `score` INT,
            `event_time` TIMESTAMP(3),
            WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
        ) WITH (
            'connector' = 'kafka',
            'topic' = 'student_topic',
            'properties.bootstrap.servers' = 'your-kafka-broker:9092',
            'properties.group.id' = 'my-group',
            'scan.startup.mode' = 'latest-offset',
            'format' = 'json'
        )
    """)

    # 建立 MySQL 結果表
    t_env.execute_sql("""
        CREATE TABLE mysql_sink (
            `id` INT,
            `name` STRING,
            `score` INT,
            PRIMARY KEY (id) NOT ENFORCED
        ) WITH (
            'connector' = 'jdbc',
            'url' = 'jdbc:mysql://your-mysql-host:3306/my_database',
            'table-name' = 'student',
            'username' = 'your_username',
            'password' = 'your_password'
        )
    """)

    # 篩選分數大於60的記錄並寫入 MySQL
    t_env.execute_sql("""
        INSERT INTO mysql_sink
        SELECT id, name, score
        FROM kafka_source
        WHERE score >= 60
    """)


if __name__ == '__main__':
    kafka_to_mysql()

預裝軟體包列表

相關文檔

• Flink Python作業的完整開發流程樣本，請參見Flink Python作業。

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

• Realtime ComputeFlink版還支援運行SQL和DataStream作業，開發方法請參見作業開發地圖和JAR作業開發。