如何使用 Flink 实现不同数据源之间的流式连接？

如何解决如何使用 Flink 实现不同数据源之间的流式连接？

我的数据来自两个不同的 Kafka 主题，由不同的代理提供服务，每个主题具有不同数量的分区。一个流包含有关正在投放的广告的事件，另一个流包含点击：

ad_serves: ad_id,ip,sTime
ad_clicks: ad_id,cTime

流程函数的文档包含一个关于使用 CoProcessFunction 或 KeyedCoProcessFunction 实现 low-level joins 的部分，但我不确定如何设置。

我也想知道这里是否可以使用 Flink 的 SQL Joins 之一。我对简单的连接都感兴趣，比如

SELECT s.ad_id,s.sTime,c.cTime
FROM ad_serves s,ad_clicks c
WHERE s.ad_id = c.ad_id

以及基于在投放后 5 秒内点击的广告的分析查询：

SELECT s.ad_id
FROM ad_serves s,ad_clicks c
WHERE
    s.ad_id = c.ad_id AND       
    s.ip = c.ip AND
    c.cTime BETWEEN s.sTime AND 
                    s.sTime + INTERVAL ‘5’ SECOND;

解决方法

总的来说，我推荐使用 Flink SQL 来实现连接，因为它易于使用且优化良好。但是无论您是使用 SQL/Table API，还是使用 DataStream API 自己实现连接，大图都大致相同。

• 您将从单独的 FlinkKafkaConsumer 来源开始，每个主题都有一个来源。如果这些主题中的分区数量（及其数据量）非常不同，那么您可能会决定相应地扩展 Flink 源的实例数量。在下图中，我通过显示 2 个 ad_serve 实例和 1 个 ad_click 实例来建议这一点。

• 在实现连接时，无论是使用 KeyedCoProcessFunction 还是 SQL/Table API，您都必须对来自两个流的键具有相等约束。在这种情况下，我们可以通过 ad_id 键控两个流。这将具有将来自两个流的所有事件的给定键的所有事件的效果——例如，下图显示了广告 17 的 ad_serve 和 ad_click 事件，以及这些事件将如何找到通往 {{1 }}。

• 作为示例给出的两个查询在必须保持多少状态方面有非常不同的要求。对于无约束的常规连接，例如

  KeyedCoProcessFunction

  执行此查询的作业必须永远存储（在 Flink 的托管键控状态下）来自两个流的所有事件。

  另一方面，第二个查询中提供的时间约束使得可以从不能再参与产生新连接结果的状态较旧的服务和点击事件中过期。 （这里我假设所涉及的流是仅附加流，其中事件大致按时间顺序排列。）

这两个查询对keying也有不同的需求。第一个查询在 SELECT s.ad_id,s.sTime,c.cTime FROM ad_serves s,ad_clicks c WHERE s.ad_id = c.ad_id 上连接； c.ad_id = s.ad_id 上的第二个。如果您想为 s.ad_id = c.ad_id AND s.ip = c.ip 设置此代码，则代码如下所示：

KeyedCoProcessFunction

请注意，连接流上的 keyBy 需要两个键选择器函数，每个流一个，并且必须将两个流映射到相同的键空间。在第二个连接的情况下，我们使用 DataStream<Serve> serves = ... DataStream<Click> clicks = ... serves .connect(clicks) .keyBy(s -> new Tuple2<>(s.ad_id,s.ip),c -> new Tuple2<>(c.ad_id,c.ip)) .process(new MyJoinFunction()) 元组作为键。