在Flink中连接多个表格时，结果不一致

Question

我们有4个定义的CDC来源，我们需要将数据组合到一个结果表中。我们使用SQL API为每个源创建一个表：例如：

"CREATE TABLE IF NOT EXISTS PAA31 (\n" +
"    WRK_SDL_DEF_NO     STRING,\n" +
"    HTR_FROM_DT        BIGINT,\n" +
...
"    update_time        TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,\n" +
"    PRIMARY KEY (WRK_SDL_DEF_NO) NOT ENFORCED,\n" +
"    WATERMARK FOR update_time AS update_time\n" +
") WITH ('value.format' = 'debezium-json' ... )";

在定义每个表之后，我们通过运行以下查询创建一个新表：

"SELECT PAA30.WRK_SDL_DEF_NO as id,\n" +
"       PAA33.DSB_TX as description,\n" +
...
"FROM PAA30\n" +
"INNER JOIN PAA33 ON PAA30.WRK_SDL_DEF_NO = PAA33.WRK_SDL_DEF_NO AND PAA33.LGG_CD = 'NL' \n" +
"INNER JOIN PAA31 ON PAA30.WRK_SDL_DEF_NO = PAA31.WRK_SDL_DEF_NO \n" +
"INNER JOIN PAA32 ON PAA30.WRK_SDL_DEF_NO = PAA32.WRK_SDL_DEF_NO";

请注意，出于格式化原因，已将一些行被排除在外。

我们遇到的问题是，执行此确切的作业会导致不一致的结果，其中有时我们有1750个产生的行（正确），但是大多数情况下，最终的行较少和随机。

这是Flink工作的计划概述。从源发送的记录数量都是正确的，但是第一个加入语句发送的记录量不是：

flink flink opercution执行计划和数字

是什么原因以及我们如何保持一致的所有数据源？

Answer 1

我看到您的管道包含一个带有处理时间触发器的事件时间窗口，并且对零顺序事件的零公差进行水印。这些可能导致问题。

Flink只能为流媒体工作负载产生完全正确的确定性结果，如果没有较晚事件，则涉及事件时间逻辑。每当处理时间逻辑干扰水印时，

• 不考虑实际的端外
• 如果使用水印发生器的配置不正确，并且如果使用闲置检测，并且
• ，则可能会发生后期事件但是，重新启动（或恢复或重新恢复）发生后

，就猜测了。将需要查看更多细节，以提供更明智的答案。最小，可再现的例子是理想的。

更新：

流媒体作业也不会散发出他们的最后一组结果，除非采取了一些挑衅。在这种情况下，例如，您可以用来

./bin/flink stop $JOB_ID --drain --savepointPath /tmp/flink-savepoints

迫使发出大型水印，以关闭最后一个窗口。

更新2：

常规加入不会产生时间属性或水印的结果。这是因为不可能保证将以任何特定的顺序排放结果，因此不可能有意义的水印。通常，在加入之后，不可能应用事件时间窗口。

更新3：

现在研究了最新代码，这显然与水印无关。

如果我正确理解的话，问题是，尽管结果始终包括应生产的内容，但额外的输出记录数量有所不同。我可以提出两个可能的原因：

（1）当flink与Debezium Server一起使用时，有可能重复的事件。我认为这不是解释，但这是值得注意的。

（2）联接的结果是非确定性的（随着运行而变化）。之所以发生这种情况，是因为各种输入流相互竞争，而摄入不同流的相关事件的确切顺序正在影响结果的产生方式。

联接的结果是一个变形流。我怀疑当结果完美时，不会发生缩回，而在其他情况下，产生了一些初步结果，这些结果后来进行了更新。

如果您检查输出流中的ROW_KIND信息，则应该能够确认此猜测是否正确。

我对Pulsar连接器并不熟悉，但我猜您应该使用UPSERT_PULSAR接收器。

Answer 2

通过启用 minibatch聚合

configuration.setString("table.exec.mini-batch.enabled", "true"); 
configuration.setString("table.exec.mini-batch.allow-latency", "500 ms"); 
configuration.setString("table.exec.mini-batch.size", "5000");

这似乎可以解决本地文件系统连接器以及Flink Pulsar Connector的一致性问题。

从这些发现中，Flink似乎在我们的吞吐量中遇到了州管理的开销问题。我们仍然需要评估现实的CDC初始负载处理，但是到目前为止，Minibatch聚合似乎有望

感谢 @David-Anderson与我们一起思考并试图弄清楚这一点。