实时计算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. 登录实时计算控制台。

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

作业调试

import logging

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

本地调试

由于实时计算Flink版默认不具备访问公网的能力，您的代码可能无法在本地直接连接线上数据源进行测试。建议按照以下方式进行本地调试：

• 单元测试：对自定义函数（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作业开发完成后，需要上传到实时计算控制台进行部署。操作步骤如下：

1. 登录实时计算控制台，进入目标工作空间。

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. 单击部署。

更多部署参数详情请参见部署作业。