0

Flink 文档中已经提到，DataSet API 将来会被弃用。因此，我正在研究以批处理模式（我相信现在处于 Beta 版）迁移的Dataset API到DataStream API 的原型。

我们的代码库中有这个（类似的）代码，它在数据集上使用leftOuterJoin。

 DataSet<SomeOutType> joined_out =  datasetA.
                leftOuterJoin(datasetB, JoinOperatorBase.JoinHint.BROADCAST_HASH_SECOND)
                .where((left) -> coalesce(left.getId(), -9999999L))
                .equalTo((right) -> right.company_id).with((JoinFunction<SomeTypeA, SomeTypeB, SomeOutType>) (left, right) -> {
                    SomeOutType recNew = SomeOutType.newBuilder().build();
                    recNew.setCustomerId(left.getCustomerId());
                    recNew.setCustomerName((right != null && right.cust_name != null) ? right.cust_name : "Blank");
                    ....
                    ....
                    ....
                    return  recNew;

                });

问题是我无法在 Datastream API docs - Join中找到Left Join或Left Outer Join等效项。

由于他们正在考虑完全弃用 DataSet API，我假设现在应该有一种方法可以在 DataStream API 中执行此 Left Outer Join。

有人可以指导我以正确的方式做到这一点吗？TIA

4

1 回答 1

0

DataSet 上的关系操作（例如，连接）已被弃用，取而代之的是使用 Table/SQL API 提供的关系操作，它与 DataStream API 完全可互操作。

请参阅https://ci.apache.org/projects/flink/flink-docs-stable/docs/dev/table/tableapi/#joins和https://ci.apache.org/projects/flink/flink-docs-稳定的/docs/dev/table/sql/queries/joins/。

您可以使用表连接器并直接在它们生成的表上执行连接，或者在执行连接之前将数据流转换为表。如果需要进一步处理，您可以从表转换回数据流。鉴于表/流的二元性，这些“转换”实际上并不需要任何成本。请参阅https://ci.apache.org/projects/flink/flink-docs-stable/docs/dev/table/data_stream_api/。

FWIW，Flink 1.14 刚刚发布，其中包含许多与该主题相关的改进。特别是，只有在 1.14（及更高版本）中，您才能将 Table API 与 DataStream API 以批处理执行模式结合起来。

于 2021-09-30T17:58:54.687 回答