PyFlink Table API 读懂 Changelog、Table API 与 SQL 混用、结果输出与 EXPLAIN 计划
1. 先把 print sink 的 Changelog 看懂
printconnector 输出的每一行，本质上是 Flink 发给 sink 的变更日志（changelog）：
格式是：
{subtask id}> {message type}{row}比如：
2> +I(4,11)
来自第 2 个并行子任务（subtask=2），+I表示 Insert（插入），行内容是(4, 11)。
1.1 三种最常见的消息类型
+I：insert（插入一条新记录）-U：update-before（撤回/删除旧值，也可理解为 retract）+U：update-after（更新后的新值）
当你执行聚合（GROUP BY、SUM、COUNT）这类会随着输入变化而不断更新结果的算子时，输出通常就不是“追加流（append-only）”，而是“更新流（updating changelog）”。
1.2 如何从 changelog 还原最终结果
看你那段输出：
6> +I(2,8) 6> -U(2,8) 6> +U(2,15)意思就是：key=2 的聚合结果先产生了(2,8)，后来被撤回（-U），并更新成(2,15)（+U）。
所以最终静态视角的结果是：
(4, 11) (2, 15) (3, 19)如果你的下游 sink 不理解-U/+U（比如某些只支持 append 的文件 sink），你就会遇到“写不进去”或“结果错乱”。这也是为什么实际落地时要考虑 sink 的 changelog 支持能力（Upsert Kafka、数据库 upsert、支持 retract 的 connector 等）。
2. Table API 和 SQL 混用：两者可以自由互转
PyFlink 的一个爽点就是：Table API 的Table和 SQL 里的表/视图是互通的。
2.1 把 Table API 的 Table 变成 SQL 可用的 view
典型用法：
# Table API 构造一张表table=table_env.from_elements([(1,'Hi'),(2,'Hello')],['id','data'])# 注册成临时视图，给 SQL 用table_env.create_temporary_view('table_api_table',table)# SQL 直接读这个 view 写入 sinktable_env.execute_sql("INSERT INTO table_sink SELECT * FROM table_api_table").wait()输出：
6> +I(1,Hi) 6> +I(2,Hello)这种写法特别适合：你想用 Table API 做一些 Python 侧拼装/条件判断，然后把主逻辑交给 SQL 写得更清晰。
2.2 把 SQL 里创建的表拿到 Table API 里做处理
SQL 先建表（例如 datagen 造数）：
table_env.execute_sql(""" CREATE TABLE sql_source ( id BIGINT, data TINYINT ) WITH ( 'connector' = 'datagen', 'fields.id.kind'='sequence', 'fields.id.start'='1', 'fields.id.end'='4', 'fields.data.kind'='sequence', 'fields.data.start'='4', 'fields.data.end'='7' ) """)再用 Table API 拿到它：
table=table_env.from_path("sql_source")table.execute().print()你会看到带op列的输出（+I）：
+----+----+------+ | op | id | data | +----+----+------+ | +I | 1 | 4 | | +I | 2 | 5 | | +I | 3 | 6 | | +I | 4 | 7 | +----+----+------+这在“SQL 建表 + Table API 做复杂处理”的组合里很常用。
3. 结果怎么拿：print / collect / pandas / 写 sink / 多 sink
很多人卡在“我算完了，怎么把结果看见/拿到手/落盘？”这里给你一张清单。
3.1 TableResult.print：打印预览（会触发物化）
table_result=table_env.execute_sql("select a + 1, b, c from %s"%source)table_result.print()注意点：这会把结果拉到客户端内存并打印，数据量大时很危险。建议配合limit控制行数：
source.limit(100)3.2 TableResult.collect：拉到客户端迭代处理（同样会物化）
table_result=table_env.execute_sql("select ...")withtable_result.collect()asresults:forrowinresults:print(row)适合：调试、小结果集、或者你要把结果继续喂给本地 Python 逻辑。
3.3 Table.to_pandas：直接转 DataFrame（更危险，慎用）
table=table_env.from_elements([(1,'Hi'),(2,'Hello')],['id','data'])print(table.to_pandas())同样会把数据拉到客户端，适合小结果预览，不适合生产跑大表。
3.4 execute_insert：写入一个 sink（最常用的“正经落地”方式）
table_env.execute_sql(""" CREATE TABLE sink_table ( id BIGINT, data VARCHAR ) WITH ('connector' = 'print') """)table=table_env.from_elements([(1,'Hi'),(2,'Hello')],['id','data'])table.execute_insert("sink_table").wait()如果你想用 SQL 写入也可以：
table_env.create_temporary_view("table_source",table)table_env.execute_sql("INSERT INTO sink_table SELECT * FROM table_source").wait()3.5 StatementSet：一个作业写多个 sink（真实项目非常常见）
当你既要落 ES，又要落 Kafka，又要打印 debug，别拆成多个 job，一个 StatementSet 就搞定：
statement_set=table_env.create_statement_set()statement_set.add_insert("first_sink_table",table1.where(col("data").like('H%')))statement_set.add_insert_sql("INSERT INTO second_sink_table SELECT * FROM simple_source")statement_set.execute().wait()好处：同一份 source/计算链路可复用，提交一个 job，运维更简单。
4. 用 EXPLAIN 看懂 Flink 到底怎么跑的
调优、排查性能瓶颈、确认下推是否生效，最终都要回到执行计划。
4.1 Table.explain：看单个 Table 的 AST / 优化计划 / 物理计划
table=table1.where(col("data").like('H%')).union_all(table2)print(table.explain())你会看到三段：
- Abstract Syntax Tree：未优化的逻辑树
- Optimized Logical Plan：优化后的逻辑计划（比如投影裁剪、谓词下推等）
- Physical Execution Plan：物理执行图（stage、算子、ship strategy 等）
排查时重点看两点：
- Filter/Projection 有没有被下推（有没有从上游算子“消失”）
- Join/Aggregation 的分发策略（hash shuffle / forward）是否符合预期
4.2 StatementSet.explain：看“多 sink 作业”的整体计划
当你用 StatementSet 同时写多个 sink 时，statement_set.explain()是最直观的全局视角，能看见每个 sink 对应的逻辑与物理落点。
5. 实战踩坑提醒
1）为什么会出现 -U/+U
只要你输出不是 append-only（典型就是聚合、去重、TopN、某些 join），print sink 就会吐 changelog。下游如果是文件、普通日志，不支持 retract 就要小心。
2）print 输出前面的X>是并行度信息
这行是哪个 subtask 打出来的，不是数据本身。并行度开大了就会多 subtask 输出混在一起。
3）collect/to_pandas/print 会把结果拉到客户端
调试 OK，生产禁用；至少先limit。
4）.wait()的使用场景
本地 mini cluster 调试用.wait()很方便；提交到远端集群时，很多场景你不希望客户端阻塞等待（尤其是流作业本来就不结束），要按你的提交方式选择是否 wait。