PyFlink打印Table结果时op列含义及查询结果去重方法咨询

op列含义 #

op列是Flink流处理模式下动态表的标准变更标记字段，用来标注当前行的changelog操作类型，无业务含义，常见取值对应逻辑如下：

• +I：新增数据行，对应数据首次进入结果集
• -U：更新前旧值行，对应数据发生更新时，被替换掉的旧版本数据
• +U：更新后新值行，对应数据发生更新后，写入结果集的新版本数据
• -D：删除数据行，对应数据从结果集中被移除
  你看到的"重复行"并非业务数据真的重复，是同一条业务数据触发更新时，Flink会按顺序先输出旧值行（带-U标记）再输出新值行（带+U标记），如果直接全量打印所有changelog行，就会出现多条同主键数据的错觉。

输出无重复结果的操作方法 #

根据执行场景选对应方案即可：

1. 有界数据场景直接用批模式执行 #

如果处理的是有界数据集（比如离线文件、全量数据库表），直接将执行环境切换为批模式，计算时只会输出最终计算结果，不会产生中间changelog，自然不会带op列、也不会出现重复行，配置代码如下：

from pyflink.table import TableEnvironment, EnvironmentSettings

# 初始化批模式执行环境
settings = EnvironmentSettings.in_batch_mode()
t_env = TableEnvironment.create(settings)

2. 无界流场景物化changelog结果 #

如果是处理实时无界流，可根据输出目的选对应处理方式：

• 本地调试查看结果：不要直接用默认print连接器打印流结果，可使用table_result.to_pandas()拉取触发计算后的最终态结果打印，或配置print连接器为upsert模式，自动合并同主键的变更行
• 结果写入外部存储：选择支持upsert语义的存储连接器（支持主键更新的JDBC连接器、HBase、Upsert Kafka等），下游存储会根据主键自动覆盖旧值，最终只保留最新版本数据，不会出现重复
• SQL层直接过滤中间值：如果需要在SQL计算链路中直接拿到去重结果，可通过业务主键做LAST_VALUE聚合、或追加窗口计算逻辑，仅保留每个主键对应的最新版本数据

注意：流模式下默认的print连接器是append模式，会原样输出所有changelog事件，不会做结果合并，直接打印必然会看到带op列的多版本行，属于正常现象不是计算逻辑错误。

内容的提问来源于stack exchange，提问作者chris