news 2026/7/16 13:04:33

PyFlink Connectors 如何在 Python 作业里正确使用 Kafka/JSON 等连接器（JAR 依赖、DDL 建表、pipeline.jars、内置 Source/Sink、

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
PyFlink Connectors 如何在 Python 作业里正确使用 Kafka/JSON 等连接器（JAR 依赖、DDL 建表、pipeline.jars、内置 Source/Sink、

1. PyFlink 为什么要手动指定 Connector/Format JAR？

因为：

  • Flink 核心运行时在 JVM 上
  • connector（如 kafka）和 format（如 json）都是 JVM 侧实现
  • Python 代码只是驱动 Table/SQL 的规划与提交

所以你需要通过pipeline.jars指定依赖（多个 jar 用;分隔）：

table_env.get_config().set("pipeline.jars","file:///my/jar/path/connector.jar;file:///my/jar/path/json.jar")

实战建议：

  • connector jar 和 format jar 都要带上（例如 Kafka + JSON）
  • 路径用file:///这种绝对 URI，避免分布式环境找不到文件
  • 生产上更推荐把 jar 放到统一位置（Flink lib 或制品仓）并在提交时声明依赖，pipeline.jars适合快速验证与 demo

2. 在 PyFlink Table API 中，推荐用 DDL 定义 Source/Sink

PyFlink 的 Table API 使用 connector 最推荐的方式是：DDL + execute_sql()
理由很简单：DDL 更直观、更可复制、也最接近线上 SQL Gateway/SQL Client 的使用方式。

2.1 Kafka Source/Sink + JSON Format（最小可用示例）

source_ddl=""" CREATE TABLE source_table( a VARCHAR, b INT ) WITH ( 'connector' = 'kafka', 'topic' = 'source_topic', 'properties.bootstrap.servers' = 'kafka:9092', 'properties.group.id' = 'test_3', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ) """sink_ddl=""" CREATE TABLE sink_table( a VARCHAR ) WITH ( 'connector' = 'kafka', 'topic' = 'sink_topic', 'properties.bootstrap.servers' = 'kafka:9092', 'format' = 'json' ) """t_env.execute_sql(source_ddl)t_env.execute_sql(sink_ddl)t_env.sql_query("SELECT a FROM source_table")\.execute_insert("sink_table").wait()

关键点拆解：

  • execute_sql()注册表（source/sink）
  • sql_query()产出一个 Table
  • execute_insert()触发写入（并提交作业）
  • .wait()在本地/mini cluster 场景常用，用于等待作业执行（远程集群通常不建议一直 wait）

3. 完整可运行的 Python 结构（把 jar、DDL、DML 串起来）

你给的完整示例结构非常标准，我建议你在博客里也用这种方式组织代码：

frompyflink.tableimportTableEnvironment,EnvironmentSettingsdeflog_processing():env_settings=EnvironmentSettings.in_streaming_mode()t_env=TableEnvironment.create(env_settings)# 1) 指定 connector & format jarst_env.get_config().set("pipeline.jars","file:///my/jar/path/connector.jar;file:///my/jar/path/json.jar")# 2) DDL: source/sinksource_ddl=""" CREATE TABLE source_table( a VARCHAR, b INT ) WITH ( 'connector' = 'kafka', 'topic' = 'source_topic', 'properties.bootstrap.servers' = 'kafka:9092', 'properties.group.id' = 'test_3', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ) """sink_ddl=""" CREATE TABLE sink_table( a VARCHAR ) WITH ( 'connector' = 'kafka', 'topic' = 'sink_topic', 'properties.bootstrap.servers' = 'kafka:9092', 'format' = 'json' ) """t_env.execute_sql(source_ddl)t_env.execute_sql(sink_ddl)# 3) DML: query + insertt_env.sql_query("SELECT a FROM source_table")\.execute_insert("sink_table")\.wait()if__name__=='__main__':log_processing()

4. PyFlink 里“内置”的 Sources/Sinks：不用额外 jar 也能跑

除了 Kafka 这类外部 connector，Flink 也提供了一些“开箱即用”的数据源/数据汇，特别适合本地调试与单测。

4.1 from/to Pandas（非常适合快速验证）

frompyflink.table.expressionsimportcolimportpandasaspdimportnumpyasnp pdf=pd.DataFrame(np.random.rand(1000,2))table=t_env.from_pandas(pdf,["a","b"]).filter(col('a')>0.5)pdf2=table.to_pandas()

注意：to_pandas()会把结果收集到客户端内存，生产慎用，建议先limit()。

4.2 from_elements()：用 Python 集合直接造表

frompyflink.tableimportDataTypes# 自动推断table_env.from_elements([(1,'Hi'),(2,'Hello')])# 指定字段名table_env.from_elements([(1,'Hi'),(2,'Hello')],['a','b'])# 指定 schema（更稳）table_env.from_elements([(1,'Hi'),(2,'Hello')],DataTypes.ROW([DataTypes.FIELD("a",DataTypes.INT()),DataTypes.FIELD("b",DataTypes.STRING())]))

这类内置 source 对写教程、做 POC、复现 bug 特别省事。

5. 自定义 Sources & Sinks：Python 不能直接写，需 Java/Scala 实现

文档明确说明了现阶段的边界：

  • 自定义 source/sink 需要 Java/Scala 实现
  • Python 侧可以通过实现 TableFactory（也是 Java/Scala）让它能被 DDL 发现并使用

也就是说：你可以用 PyFlink 写作业逻辑，但 connector 生态仍然是 JVM 的。

如果你后面要写“自定义 connector”系列博客，可以按这个路线写：

  • 先用 Java 写 DynamicTableSourceFactory / DynamicTableSinkFactory（SPI 注册）
  • 再在 PyFlink 里通过 DDL'connector'='xxx'直接使用

6. 常见踩坑清单（PyFlink Connector 场景高频问题）

  • 只加了 connector jar，没加 format jar：DDL 里用了'format'='json'，但没带 json format 的 jar，会在运行期报找不到 format factory
  • pipeline.jars 路径不可达：本地 file 路径对集群 TaskManager 不可见，必须用集群可访问路径或随 job 提交
  • 用 DDL 建表但没触发执行：Table/SQL 是惰性执行，必须execute_insert()或execute_sql(INSERT ...)才会提交作业
  • wait() 用错场景：本地调试很方便；远程集群提交通常希望异步返回，避免客户端阻塞
版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/7/15 22:17:31

LLaMA Factory+云端GPU：毕业设计救星，快速搞定AI项目

LLaMA Factory云端GPU：毕业设计救星，快速搞定AI项目 临近毕业季，计算机专业的学生小李急需一个强大的GPU环境来完成他的大模型相关毕业设计，但学校服务器需要排队两周以上。如果你也面临类似困境，LLaMA Factory结合云…

作者头像 李华
网站建设 2026/7/15 22:17:29

基于YOLOv8/YOLOv7/YOLOv6/YOLOv5的危险物品检测系统（深度学习模型+PySide6界面+训练数据集+Python代码）

摘要 随着公共安全需求的日益增长，危险物品检测技术在社会安防、交通安检等领域发挥着重要作用。本文介绍了一个基于YOLO系列深度学习框架的危险物品检测系统，集成了YOLOv8、YOLOv7、YOLOv6和YOLOv5四种先进的物体检测算法。系统采用PySide6开发了用户友好的图形界面，提供了…

作者头像 李华
网站建设 2026/7/14 14:41:27

用JADX快速验证APP创意：1小时完成竞品分析原型

快速体验 打开 InsCode(快马)平台 https://www.inscode.net输入框内输入如下内容： 创建一个竞品快速分析工具原型，基于JADX实现以下功能：1. 自动提取竞品核心功能模块；2. 对比多个APK的架构差异；3. 生成竞争力分析报…

作者头像 李华
网站建设 2026/7/14 14:40:29

CLAUDE CODE收费模式如何提升开发效率？

快速体验 打开 InsCode(快马)平台 https://www.inscode.net输入框内输入如下内容： 开发一个效率-成本计算器，功能包括：1) 工时输入界面 2) 传统开发与AI辅助开发效率对比 3) 实时成本差异计算 4) ROI分析图表。要求使用Vue.js构建响应式界面…

作者头像 李华
网站建设 2026/7/14 14:41:20

Llama Factory对比测试：不同量化方法对模型效果的影响一目了然

Llama Factory对比测试：不同量化方法对模型效果的影响一目了然 作为一名移动端开发者，你是否遇到过这样的困扰：想把大模型部署到App中，但面对五花八门的量化方法却无从下手？GPTQ、AWQ、Bitsandbytes...这些量化策略到底…

作者头像 李华
网站建设 2026/6/29 0:09:50

安全第一：在隔离环境中用Llama Factory微调敏感数据模型

安全第一：在隔离环境中用Llama Factory微调敏感数据模型 医疗数据的安全性是AI开发中的首要考量。本文将详细介绍如何在完全隔离的环境中，使用Llama Factory工具对敏感医疗数据进行安全微调，确保患者隐私不被泄露。 为什么需要隔离环境&#…

作者头像 李华