使用线程在消费者流中循环运行 Flink 消费者和 Flink 生产者

Question

我想让 flink 消费者流中的每条消息产生多条消息，每条消息都通过一个单独的线程，使用 flink kafka 生产者发送到 kafka 中的某个主题。我用 Scala 编写程序，但用 Java 回答就可以了

类似这样的：

def thread(x:String): Thread =
    {   
    val thread_ = new Thread {
          override def run {

              val str = some_processing(x)
              flink_producer(str)

      }
    }
    return thread_
 }

val stream = flink_consumer()

stream.map(x =>{

                 var i = 0
                 while(i < 10){

                                val th = thread(x)
                                th.start()
                                i = i+1
                                }

           })

因此，对于 flink 消费者中的每个输入，我想使用多线程向其他队列生成 10 条消息。

Answer 1

大多数 Flink 算子都是并行算子，因此您没有理由在数据管道中创建任何类型的线程，Flink 应该是管理算子可以存在多少并行实例以及是否要设置该值，您应该使用以下 API 方法。

.setParallelism(N) //N is 10 for you,

你可以得到more info in Fink documentation

你应该这样做：

1. 为您的集群配置添加更多任务管理器插槽
2. 使用生成 10 条消息的 flatMap 代替地图
3. 将 flatMap 运算符的并行度增加到 10。

您的代码应如下所示：

val stream = flink_consumer()

stream.flatMap((x, out) =>{
                 var i = 0
                 while(i < 10){
                      val valueToCollect = process(x,i)
                      out.collect(valueToCollect)
                 }

           }).setParallelism(10)
           .map(doSomethingWithGeneratedValues)
           .addSink(sinkThatSendsDataToYourDesiredSystem)

另一种方法，如果你知道你想要多少并行任务

val stream = flink_consumer()

val resultStream = stream.map(process)
val sinkStream = resultStream.union(resultStream,resultStream,resultStream,...) // joins resultStream N times
sinkStream.addSink(sinkThatSendsDataToYourDesiredSystem)

最后，您还可以为一个 DataStream 设置多个接收器

val stream = flink_consumer()

val resultStream = stream.map(process)
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)
...
N
...
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)

如果您想对数据接收器进行并行写入，则必须确保您使用的接收器支持这种写入操作。