0

I would like to have each message in the flink consumer stream produce multiple messages, each via a seperate thread, to some topic in kafka using flink kafka producer.Im writing the program in Scala but answers in Java will do

Something like this:

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
                                }

           })

So for each input in flink consumer I would like to produce 10 messages to some other queue using multi-threading.

1 Answer 1

1

Most of Flink operators are parallel operators, so there's no reason for you to create any kind of thread in your data pipeline, Flink should be the one that manages how many parallel instances could exist of an operator and if you want to set that value, you should use the following API method.

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

You can get more info in Fink documentation

You sould do something like this:

  1. add more task managers slots to your cluster config
  2. instead of map use a flatMap that generates thoose 10 messages
  3. increase the parallelism of the flatMap operator to 10.

Your code should look like this:

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)

Another aproach if you know how many parallel task you want to have

val stream = flink_consumer()

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

Finally, you can also have multiple sinks for a DataStream

val stream = flink_consumer()

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

If you want to do parallel writes to your data sink, you must ensure that the sink you use has support to that kind of write operations.

9
  • Thank you for the answer. According to my understanding, the flatmap would take one consumed message and produce 10 unique mesages with the help of process(x,i) and then send 10 messages to kafka using addsink.I would get 10 messages in the Kafka queue from the one message.So what is the use of map over here?
    – Rolf Lobo
    Feb 5, 2018 at 12:50
  • Yes, you are right, I used the flatMap because i thought that you want to generate a new message per Thread, if you only want to process data in parallel, a simple map will do the job ;)
    – diegoreico
    Feb 6, 2018 at 8:59
  • Ok so I'll need a flatmap but is there a way to replace the while loop with a parrallel iterator?
    – Rolf Lobo
    Feb 6, 2018 at 9:45
  • You can use a scala parallel iterator, but that's gonna create Threads inside a Flink task and that doesn't look like a good practice. Because, if you have as many task slots as CPU Cores in your node, creating a lot of threads it's going to produce a lot of context switchs, so in the end it's going to be slower than a normal while loop. Anyways, i never did a benchmark about this topics, so i can't be 100% sure
    – diegoreico
    Feb 6, 2018 at 10:39
  • Another good aproach, could be the addition of multiple sinks, to a map operator. I'm going to update my answer
    – diegoreico
    Feb 6, 2018 at 10:43

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.