Flink Table API 如何将同一上游表的多查询合并为单作业执行

两个查询完全可以合并为单个作业运行。当前代码生成两个独立作业的根本原因非常明确：Table.execute()是单Pipeline触发方法，每调用一次就会基于当前Table从源端开始构建独立DAG并提交作业，两次调用之间的上下文完全隔离；哪怕共享同一个DataStream转换的临时表，也不会复用已有的计算逻辑，甚至会重复读取上游CSV文件造成资源浪费。

方案1：通过Union All合并互斥结果为单Table输出 #

当两类结果不需要分开存储、分开展示时，可以将两个过滤后的结果集合并为同一个Table，仅调用一次execute()触发单作业提交。由于两个过滤条件（贷款状态为Charged Off / Fully Paid）天然互斥，合并后不会出现重复数据，还可以通过结果自带的loanStatus字段直接区分两类记录。

改造后的核心代码逻辑示例：

tableEnv.createTemporaryView("CreditDetails", creditStream);
Table creditDetailsTable = tableEnv.from("CreditDetails");

// 保留两个过滤逻辑，不单独触发执行
Table chargedOffTable = creditDetailsTable.select($("*"))
        .filter($("loanStatus").isEqual("Charged Off"));
Table fullyPaidTable = creditDetailsTable.select($("*"))
        .filter($("loanStatus").isEqual("Fully Paid"));

// 合并两个逻辑为同一个计算Pipeline
Table mergedResult = chargedOffTable.unionAll(fullyPaidTable);

// 仅触发一次执行，生成单个作业
TableResult result = mergedResult.execute();
result.print();

这种模式下，上游的文件读取、CSV解析Map算子在整个DAG中只会存在一个实例，两个过滤分支是同一个作业内的并行算子，无重复资源开销。

方案2：通过StatementSet绑定多Sink实现独立输出 #

如果需要把两类结果输出到完全不同的下游（比如不同的Kafka Topic、不同的数据库表、独立的打印控制台），不需要把结果混在同一个输出流里，可以使用StatementSet批量收集多个Sink的插入逻辑，一次性提交所有逻辑为单作业。

核心代码逻辑示例：

// 创建用于批量收集执行逻辑的StatementSet实例
StatementSet statementSet = tableEnv.createStatementSet();

// 保留两个独立的结果查询逻辑
Table chargedOffTable = creditDetailsTable.select($("*"))
        .filter($("loanStatus").isEqual("Charged Off"));
Table fullyPaidTable = creditDetailsTable.select($("*"))
        .filter($("loanStatus").isEqual("Fully Paid"));

// 为两类结果分别创建独立的输出Sink（这里以print sink为例）
tableEnv.createTemporaryTable("ChargedOffOutput", TableDescriptor.forConnector("print").build());
tableEnv.createTemporaryTable("FullyPaidOutput", TableDescriptor.forConnector("print").build());

// 将两个结果逻辑绑定到对应Sink，加入同一次批量提交
statementSet.addInsert("ChargedOffOutput", chargedOffTable);
statementSet.addInsert("FullyPaidOutput", fullyPaidTable);

// 一次性提交所有逻辑，生成单个作业
statementSet.execute();

这种模式下两类结果可以独立输出到不同目标，同时整个计算链路的共享算子只会启动一次，不会重复消费上游DataStream数据。

• 不要在多查询的逻辑链路中提前调用Table.execute()，该方法一旦触发就会隔离当前的DAG上下文，后续的查询会从零开始构建全新的作业，无法再和之前的逻辑合并。
• 基于DataStream创建的临时表，只要没有被提前执行触发算子实例化，Blink Planner会自动在同一个Pipeline中实现上游算子的共享，不会出现重复启动Source、解析算子的问题。
• 两种合并方案在流模式下同样生效，不会因为多分支逻辑改变作业的流处理语义。

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