根据优先级或循环方式从两个flink dataStream消耗 [英] Consume from two flink dataStream based on priority or round robin way

问题描述

我有两个flink dataStream.例如:dataStream1和dataStream2.我想将两个流合并为1个流，以便可以使用相同的处理函数来处理它们，因为两个dataStream的dag相同.

I have two flink dataStream. For ex: dataStream1 and dataStream2. I want to union both the Streams into 1 stream so that I can process them using the same process functions as the dag of both dataStream is the same.

截至目前，对于任何一个流，我都需要消耗消息的优先级相同. dataStream2的生产者每分钟产生10条消息，而dataStream1的生产者每秒产生1000条消息.此外，两个dataStreams.DataSteam2的dataTypes相同，应尽快使用一个高优先级队列. dataStream1和dataStream2的消息之间没有任何关系

As of now, I need equal priority of consumption of messages for either stream. The producer of dataStream2 produces 10 messages per minute, while the producer of dataStream1 produces 1000 messages per second. Also, dataTypes are the same for both dataStreams.DataSteam2 more of a high priority queue that should be consumed asap. There is no relation between messages of dataStream1 and dataStream2

dataStream1.union(dataStream2)是否会生成同时包含两个Streams元素的Stream?

Does dataStream1.union(dataStream2) will produce a Stream that will have elements of both Streams?

推荐答案

此问题可能是最简单的解决方案，但取决于您的数据源的确切规范，并不是最有效的解决方案，可能是将两个流连接在一起.在此解决方案中，您可以使用CoProcessFunction，它将为每个连接的流调用单独的方法.

Probably the simplest solution to this problem, yet not exactly the most efficient one depending on the exact specification of the sources for Your data, may be connecting the two streams. In this solution, You could use the CoProcessFunction, which will invoke separate methods for each of the connected streams.

在此解决方案中，您可以简单地缓冲一个流的元素，直到可以生成它们为止(例如，以循环方式).但是请记住，如果源产生事件的频率之间存在很大差异，则效率可能会很低.

In this solution, You could simply buffer the elements of one stream until they can be produced (for example in round-robin manner). But keep in mind that this may be quite inefficient if there is a very big difference between the frequency in which sources produce events.

这篇关于根据优先级或循环方式从两个flink dataStream消耗的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！