执行 Apache BEAM sql 查询时出错 - 在 GroupByKey 之前使用 Window.into 或 Window.triggering 转换

Question

如何在 BEAM SQL 中的 GroupByKey 之前包含 Window.into 或 Window.triggering 转换？

我有以下 2 个表：

表

CREATE TABLE table1(
field1 varchar 
,field2 varchar
)

第二表

CREATE TABLE table2(
field1 varchar
,field3 varchar
)

我正在将结果写在第三个表中

CREATE TABLE table3(
field1 varchar
,field3 varchar
)

前 2 个表正在从 kafka 流中读取数据，我正在对这些表进行连接并将数据插入到第三个表中，使用以下查询。前 2 个表是无界/无界的

INSERT INTO table3 
        (field1, 
         field3) 
SELECT a.field1, 
   b.field3 
FROM   table1 a 
   JOIN table2 b 
     ON a.field1 = b.field1 

我收到以下错误：

引起：java.lang.IllegalStateException：GroupByKey 不能应用于没有触发器的 GlobalWindow 中的无界 PCollection。在 GroupByKey 之前使用 Window.into 或 Window.triggering 转换。在 org.apache.beam.sdk.transforms.GroupByKey.applicableTo(GroupByKey.java:173) 在 org.apache.beam.sdk.transforms.GroupByKey.expand(GroupByKey.java:204) 在 org.apache.beam.sdk .transforms.GroupByKey.expand(GroupByKey.java:120) 在 org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:537) 在 org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:472) ) 在 org.apache.beam.sdk.values.PCollection.apply(PCollection.java:286) 在 org.apache.beam.sdk.transforms.join.CoGroupByKey.expand(CoGroupByKey.java:126) 在 org.apache。 org.beam.sdk.transforms.join.CoGroupByKey.expand(CoGroupByKey.java:74)。

Answer 1

这是 Beam SQL 的当前实现限制。您需要定义 windows然后加入输入 per-window。

关于如何在 Beam SQL 中进行连接和窗口化的几个示例：

• 带有HOP窗口和连接的复杂 SQL 查询；
• 测试，它在 SQL 之外用 Java 定义一个窗口，然后应用带有连接的查询；
• 其他窗口函数的例子可以在这里找到；

背景

问题是由于一般情况下很难定义无界数据流的Join操作，它不仅限于Beam SQL。

例如，想象一下，当数据处理系统从 2 个来源接收输入，然后必须匹配它们之间的记录时。从高层次的角度来看，这样的系统必须保留它迄今为止看到的所有数据，然后对于每个新记录，它必须遍历第二个输入源中的所有记录，以查看那里是否存在匹配项。当您拥有有限且小型的数据源时，它可以正常工作。在简单的情况下，您可以将所有内容加载到内存中，匹配来自源的数据，生成输出。

使用流数据，你不能永远缓存它。如果数据永远不会停止会怎样？并且不清楚何时要发出数据。如果你有一个outer join操作，你什么时候决定你没有来自另一个输入的匹配记录？

例如，请参阅Beam 指南中对 unbounded in部分的解释。并且 Beam 中的连接通常使用（Beam SQL Joins 以及）在它之上实现。PCollectionsGroupByKeyCoGroupByKey

对于特定的管道，所有这些问题都可能得到解答，但在一般情况下很难解决。目前 Beam SDK 和 Beam SQL 的方法是将其委托给用户来解决具体的业务案例。Beam 允许用户决定将哪些数据聚合到一个窗口中、等待延迟数据的时间以及何时发出结果。还有诸如状态单元和计时器之类的东西，可以进行更精细的控制。这允许编写管道的程序员显式定义行为并在一定程度上解决这些问题，但具有（很多）额外的复杂性。

Beam SQL 是在常规 Beam SDK 概念之上实现的，并受到相同限制的约束。但它有更多自己的实现。例如，您没有 SQL 语法来定义触发器、状态或自定义窗口。或者您不能编写ParDo可以在外部服务中保持状态的自定义。