You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by "r. r." <ro...@abv.bg> on 2017/10/02 16:46:27 UTC
kafka consumer parallelism
Hello
I'm running a job with "flink run -p5" and additionally set env.setParallelism(5).
The source of the stream is Kafka, the job uses FlinkKafkaConsumer010.
In Flink UI though I notice that if I send 3 documents to Kafka, only one 'instance' of the consumer seems to receive Kafka's record and send them to next operators, which according to Flink UI are properly parallelized.
What's the explanation of this behavior?
According to sources:
To enable parallel execution, the user defined source should
* implement {@link org.apache.flink.streaming.api.functions.source.ParallelSourceFunction} or extend {@link
* org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction}
which FlinkKafkaConsumer010 does
Please check a screenshot at https://imgur.com/a/E1H9r you'll see that only one sends 3 records to the sinks
My code is here: https://pastebin.com/yjYCXAAR
Thanks!
Re: kafka consumer parallelism
Posted by "r. r." <ro...@abv.bg>.
Thanks a lot, Carst!
I hadn't realized that
Best regards
>-------- Оригинално писмо --------
>От: Carst Tankink ctankink@bol.com
>Относно: Re: kafka consumer parallelism
>До: "r. r." <ro...@abv.bg>
>Изпратено на: 05.10.2017 09:04
> Hi,
>
> The latter (map will be spread out if you rebalance before it).
> You can also see it in the flink dashboard you screen-shotted: the Source and the map are in the same ‘block’, so the operators are chained to the same task (and will run at the same parallelism/slot).
>
>
> Carst
>
> On 10/4/17, 12:36, "r. r." <ro...@abv.bg> wrote:
>
> Thanks Timo & Tovi - this helped me get a better idea how it works
>
> @Carst, I have rebalance after the map() (messageStream.map(...).rebalance()) - doesn't it mean the load will be redistributed across all job managers' slots anyway?
> Or is the map() spread out only if I do as you suggest messageStream.rebalance().map(..) ?
>
> Best regards
> Rob
>
>
>
>
>
>
>
>
> >-------- Оригинално писмо --------
>
> >От: Carst Tankink ctankink@bol.com
>
> >Относно: Re: kafka consumer parallelism
>
> >До: "user@flink.apache.org" <us...@flink.apache.org>
>
> >Изпратено на: 03.10.2017 11:30
>
>
>
>
> > (Accidentally sent this to Timo instead of to-list...)
>
> >
>
> > Hi,
>
> >
>
> > What Timo says is true, but in case you have a higher parallism than the number of partitions (because you want to make use of it in a future operator), you could do a .rebalance() (see https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/datastream_api.html#physical-partitioning) after the Kafka source.
>
> > This makes sure that all operators after the Kafka source get an even load, at the cost of having to redistribute the documents (so there is de/serialization + network overhead).
>
> >
>
> >
>
> > Carst
>
> >
>
> > On 10/3/17, 09:34, "Sofer, Tovi " <to...@citi.com> wrote:
>
> >
>
> > Hi Robert,
>
> >
>
> > I had similar issue.
>
> > For me the problem was that the topic was auto created with one partition.
>
> > You can alter it to have 5 partitions using kafka-topics command.
>
> > Example:
>
> > kafka-topics --alter --partitions 5 --topic fix --zookeeper localhost:2181
>
> >
>
> > Regards,
>
> > Tovi
>
> > -----Original Message-----
>
> > From: Timo Walther [mailto:twalthr@apache.org]
>
> > Sent: יום ב 02 אוקטובר 2017 20:59
>
> > To: user@flink.apache.org
>
> > Subject: Re: kafka consumer parallelism
>
> >
>
> > Hi,
>
> >
>
> > I'm not a Kafka expert but I think you need to have more than 1 Kafka partition to process multiple documents at the same time. Make also sure to send the documents to different partitions.
>
> >
>
> > Regards,
>
> > Timo
>
> >
>
> >
>
> > Am 10/2/17 um 6:46 PM schrieb r. r.:
>
> > > Hello
>
> > > I'm running a job with "flink run -p5" and additionally set env.setParallelism(5).
>
> > > The source of the stream is Kafka, the job uses FlinkKafkaConsumer010.
>
> > > In Flink UI though I notice that if I send 3 documents to Kafka, only one 'instance' of the consumer seems to receive Kafka's record and send them to next operators, which according to Flink UI are properly parallelized.
>
> > > What's the explanation of this behavior?
>
> > > According to sources:
>
> > >
>
> > > To enable parallel execution, the user defined source should
>
> > > * implement {@link
>
> > > org.apache.flink.streaming.api.functions.source.ParallelSourceFunction
>
> > > } or extend {@link
>
> > > *
>
> > > org.apache.flink.streaming.api.functions.source.RichParallelSourceFunc
>
> > > tion}
>
> > > which FlinkKafkaConsumer010 does
>
> > >
>
> > > Please check a screenshot at
>
> > > https://urldefense.proofpoint.com/v2/url?u=https-3A__imgur.com_a_E1H9r
>
> > > &d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3rJ99
>
> > > _MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=ti6cswIJ4X9
>
> > > d5wgGkq5EUx41y4WXZ_z_HebkoOrLEmw&e= you'll see that only one sends 3
>
> > > records to the sinks
>
> > >
>
> > > My code is here:
>
> > > https://urldefense.proofpoint.com/v2/url?u=https-3A__pastebin.com_yjYC
>
> > > XAAR&d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3
>
> > > rJ99_MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=AApHKm3
>
> > > amPLzWwAqk2KITEeUkhNE0GS1Oo02jaUpKIw&e=
>
> > >
>
> > > Thanks!
>
> >
>
> >
>
> >
>
Re: kafka consumer parallelism
Posted by Carst Tankink <ct...@bol.com>.
Hi,
The latter (map will be spread out if you rebalance before it).
You can also see it in the flink dashboard you screen-shotted: the Source and the map are in the same ‘block’, so the operators are chained to the same task (and will run at the same parallelism/slot).
Carst
On 10/4/17, 12:36, "r. r." <ro...@abv.bg> wrote:
Thanks Timo & Tovi - this helped me get a better idea how it works
@Carst, I have rebalance after the map() (messageStream.map(...).rebalance()) - doesn't it mean the load will be redistributed across all job managers' slots anyway?
Or is the map() spread out only if I do as you suggest messageStream.rebalance().map(..) ?
Best regards
Rob
>-------- Оригинално писмо --------
>От: Carst Tankink ctankink@bol.com
>Относно: Re: kafka consumer parallelism
>До: "user@flink.apache.org" <us...@flink.apache.org>
>Изпратено на: 03.10.2017 11:30
> (Accidentally sent this to Timo instead of to-list...)
>
> Hi,
>
> What Timo says is true, but in case you have a higher parallism than the number of partitions (because you want to make use of it in a future operator), you could do a .rebalance() (see https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/datastream_api.html#physical-partitioning) after the Kafka source.
> This makes sure that all operators after the Kafka source get an even load, at the cost of having to redistribute the documents (so there is de/serialization + network overhead).
>
>
> Carst
>
> On 10/3/17, 09:34, "Sofer, Tovi " <to...@citi.com> wrote:
>
> Hi Robert,
>
> I had similar issue.
> For me the problem was that the topic was auto created with one partition.
> You can alter it to have 5 partitions using kafka-topics command.
> Example:
> kafka-topics --alter --partitions 5 --topic fix --zookeeper localhost:2181
>
> Regards,
> Tovi
> -----Original Message-----
> From: Timo Walther [mailto:twalthr@apache.org]
> Sent: יום ב 02 אוקטובר 2017 20:59
> To: user@flink.apache.org
> Subject: Re: kafka consumer parallelism
>
> Hi,
>
> I'm not a Kafka expert but I think you need to have more than 1 Kafka partition to process multiple documents at the same time. Make also sure to send the documents to different partitions.
>
> Regards,
> Timo
>
>
> Am 10/2/17 um 6:46 PM schrieb r. r.:
> > Hello
> > I'm running a job with "flink run -p5" and additionally set env.setParallelism(5).
> > The source of the stream is Kafka, the job uses FlinkKafkaConsumer010.
> > In Flink UI though I notice that if I send 3 documents to Kafka, only one 'instance' of the consumer seems to receive Kafka's record and send them to next operators, which according to Flink UI are properly parallelized.
> > What's the explanation of this behavior?
> > According to sources:
> >
> > To enable parallel execution, the user defined source should
> > * implement {@link
> > org.apache.flink.streaming.api.functions.source.ParallelSourceFunction
> > } or extend {@link
> > *
> > org.apache.flink.streaming.api.functions.source.RichParallelSourceFunc
> > tion}
> > which FlinkKafkaConsumer010 does
> >
> > Please check a screenshot at
> > https://urldefense.proofpoint.com/v2/url?u=https-3A__imgur.com_a_E1H9r
> > &d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3rJ99
> > _MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=ti6cswIJ4X9
> > d5wgGkq5EUx41y4WXZ_z_HebkoOrLEmw&e= you'll see that only one sends 3
> > records to the sinks
> >
> > My code is here:
> > https://urldefense.proofpoint.com/v2/url?u=https-3A__pastebin.com_yjYC
> > XAAR&d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3
> > rJ99_MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=AApHKm3
> > amPLzWwAqk2KITEeUkhNE0GS1Oo02jaUpKIw&e=
> >
> > Thanks!
>
>
>
Re: kafka consumer parallelism
Posted by "r. r." <ro...@abv.bg>.
Thanks Timo & Tovi - this helped me get a better idea how it works
@Carst, I have rebalance after the map() (messageStream.map(...).rebalance()) - doesn't it mean the load will be redistributed across all job managers' slots anyway?
Or is the map() spread out only if I do as you suggest messageStream.rebalance().map(..) ?
Best regards
Rob
>-------- Оригинално писмо --------
>От: Carst Tankink ctankink@bol.com
>Относно: Re: kafka consumer parallelism
>До: "user@flink.apache.org" <us...@flink.apache.org>
>Изпратено на: 03.10.2017 11:30
> (Accidentally sent this to Timo instead of to-list...)
>
> Hi,
>
> What Timo says is true, but in case you have a higher parallism than the number of partitions (because you want to make use of it in a future operator), you could do a .rebalance() (see https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/datastream_api.html#physical-partitioning) after the Kafka source.
> This makes sure that all operators after the Kafka source get an even load, at the cost of having to redistribute the documents (so there is de/serialization + network overhead).
>
>
> Carst
>
> On 10/3/17, 09:34, "Sofer, Tovi " <to...@citi.com> wrote:
>
> Hi Robert,
>
> I had similar issue.
> For me the problem was that the topic was auto created with one partition.
> You can alter it to have 5 partitions using kafka-topics command.
> Example:
> kafka-topics --alter --partitions 5 --topic fix --zookeeper localhost:2181
>
> Regards,
> Tovi
> -----Original Message-----
> From: Timo Walther [mailto:twalthr@apache.org]
> Sent: יום ב 02 אוקטובר 2017 20:59
> To: user@flink.apache.org
> Subject: Re: kafka consumer parallelism
>
> Hi,
>
> I'm not a Kafka expert but I think you need to have more than 1 Kafka partition to process multiple documents at the same time. Make also sure to send the documents to different partitions.
>
> Regards,
> Timo
>
>
> Am 10/2/17 um 6:46 PM schrieb r. r.:
> > Hello
> > I'm running a job with "flink run -p5" and additionally set env.setParallelism(5).
> > The source of the stream is Kafka, the job uses FlinkKafkaConsumer010.
> > In Flink UI though I notice that if I send 3 documents to Kafka, only one 'instance' of the consumer seems to receive Kafka's record and send them to next operators, which according to Flink UI are properly parallelized.
> > What's the explanation of this behavior?
> > According to sources:
> >
> > To enable parallel execution, the user defined source should
> > * implement {@link
> > org.apache.flink.streaming.api.functions.source.ParallelSourceFunction
> > } or extend {@link
> > *
> > org.apache.flink.streaming.api.functions.source.RichParallelSourceFunc
> > tion}
> > which FlinkKafkaConsumer010 does
> >
> > Please check a screenshot at
> > https://urldefense.proofpoint.com/v2/url?u=https-3A__imgur.com_a_E1H9r
> > &d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3rJ99
> > _MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=ti6cswIJ4X9
> > d5wgGkq5EUx41y4WXZ_z_HebkoOrLEmw&e= you'll see that only one sends 3
> > records to the sinks
> >
> > My code is here:
> > https://urldefense.proofpoint.com/v2/url?u=https-3A__pastebin.com_yjYC
> > XAAR&d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3
> > rJ99_MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=AApHKm3
> > amPLzWwAqk2KITEeUkhNE0GS1Oo02jaUpKIw&e=
> >
> > Thanks!
>
>
>
Re: kafka consumer parallelism
Posted by Carst Tankink <ct...@bol.com>.
(Accidentally sent this to Timo instead of to-list...)
Hi,
What Timo says is true, but in case you have a higher parallism than the number of partitions (because you want to make use of it in a future operator), you could do a .rebalance() (see https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/datastream_api.html#physical-partitioning) after the Kafka source.
This makes sure that all operators after the Kafka source get an even load, at the cost of having to redistribute the documents (so there is de/serialization + network overhead).
Carst
On 10/3/17, 09:34, "Sofer, Tovi " <to...@citi.com> wrote:
Hi Robert,
I had similar issue.
For me the problem was that the topic was auto created with one partition.
You can alter it to have 5 partitions using kafka-topics command.
Example:
kafka-topics --alter --partitions 5 --topic fix --zookeeper localhost:2181
Regards,
Tovi
-----Original Message-----
From: Timo Walther [mailto:twalthr@apache.org]
Sent: יום ב 02 אוקטובר 2017 20:59
To: user@flink.apache.org
Subject: Re: kafka consumer parallelism
Hi,
I'm not a Kafka expert but I think you need to have more than 1 Kafka partition to process multiple documents at the same time. Make also sure to send the documents to different partitions.
Regards,
Timo
Am 10/2/17 um 6:46 PM schrieb r. r.:
> Hello
> I'm running a job with "flink run -p5" and additionally set env.setParallelism(5).
> The source of the stream is Kafka, the job uses FlinkKafkaConsumer010.
> In Flink UI though I notice that if I send 3 documents to Kafka, only one 'instance' of the consumer seems to receive Kafka's record and send them to next operators, which according to Flink UI are properly parallelized.
> What's the explanation of this behavior?
> According to sources:
>
> To enable parallel execution, the user defined source should
> * implement {@link
> org.apache.flink.streaming.api.functions.source.ParallelSourceFunction
> } or extend {@link
> *
> org.apache.flink.streaming.api.functions.source.RichParallelSourceFunc
> tion}
> which FlinkKafkaConsumer010 does
>
> Please check a screenshot at
> https://urldefense.proofpoint.com/v2/url?u=https-3A__imgur.com_a_E1H9r
> &d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3rJ99
> _MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=ti6cswIJ4X9
> d5wgGkq5EUx41y4WXZ_z_HebkoOrLEmw&e= you'll see that only one sends 3
> records to the sinks
>
> My code is here:
> https://urldefense.proofpoint.com/v2/url?u=https-3A__pastebin.com_yjYC
> XAAR&d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3
> rJ99_MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=AApHKm3
> amPLzWwAqk2KITEeUkhNE0GS1Oo02jaUpKIw&e=
>
> Thanks!
RE: kafka consumer parallelism
Posted by "Sofer, Tovi " <to...@citi.com>.
Hi Robert,
I had similar issue.
For me the problem was that the topic was auto created with one partition.
You can alter it to have 5 partitions using kafka-topics command.
Example:
kafka-topics --alter --partitions 5 --topic fix --zookeeper localhost:2181
Regards,
Tovi
-----Original Message-----
From: Timo Walther [mailto:twalthr@apache.org]
Sent: יום ב 02 אוקטובר 2017 20:59
To: user@flink.apache.org
Subject: Re: kafka consumer parallelism
Hi,
I'm not a Kafka expert but I think you need to have more than 1 Kafka partition to process multiple documents at the same time. Make also sure to send the documents to different partitions.
Regards,
Timo
Am 10/2/17 um 6:46 PM schrieb r. r.:
> Hello
> I'm running a job with "flink run -p5" and additionally set env.setParallelism(5).
> The source of the stream is Kafka, the job uses FlinkKafkaConsumer010.
> In Flink UI though I notice that if I send 3 documents to Kafka, only one 'instance' of the consumer seems to receive Kafka's record and send them to next operators, which according to Flink UI are properly parallelized.
> What's the explanation of this behavior?
> According to sources:
>
> To enable parallel execution, the user defined source should
> * implement {@link
> org.apache.flink.streaming.api.functions.source.ParallelSourceFunction
> } or extend {@link
> *
> org.apache.flink.streaming.api.functions.source.RichParallelSourceFunc
> tion}
> which FlinkKafkaConsumer010 does
>
> Please check a screenshot at
> https://urldefense.proofpoint.com/v2/url?u=https-3A__imgur.com_a_E1H9r
> &d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3rJ99
> _MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=ti6cswIJ4X9
> d5wgGkq5EUx41y4WXZ_z_HebkoOrLEmw&e= you'll see that only one sends 3
> records to the sinks
>
> My code is here:
> https://urldefense.proofpoint.com/v2/url?u=https-3A__pastebin.com_yjYC
> XAAR&d=DwIDaQ&c=j-EkbjBYwkAB4f8ZbVn1Fw&r=bfLStYBPfgr58eRbGoW11gp3x4kr3
> rJ99_MiSMX5oOs&m=LiwKApZmqwYYsiKqby4Ugd5WJgyPKpj3H7s9l7Xw_Qg&s=AApHKm3
> amPLzWwAqk2KITEeUkhNE0GS1Oo02jaUpKIw&e=
>
> Thanks!
Re: kafka consumer parallelism
Posted by Timo Walther <tw...@apache.org>.
Hi,
I'm not a Kafka expert but I think you need to have more than 1 Kafka
partition to process multiple documents at the same time. Make also sure
to send the documents to different partitions.
Regards,
Timo
Am 10/2/17 um 6:46 PM schrieb r. r.:
> Hello
> I'm running a job with "flink run -p5" and additionally set env.setParallelism(5).
> The source of the stream is Kafka, the job uses FlinkKafkaConsumer010.
> In Flink UI though I notice that if I send 3 documents to Kafka, only one 'instance' of the consumer seems to receive Kafka's record and send them to next operators, which according to Flink UI are properly parallelized.
> What's the explanation of this behavior?
> According to sources:
>
> To enable parallel execution, the user defined source should
> * implement {@link org.apache.flink.streaming.api.functions.source.ParallelSourceFunction} or extend {@link
> * org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction}
> which FlinkKafkaConsumer010 does
>
> Please check a screenshot at https://imgur.com/a/E1H9r you'll see that only one sends 3 records to the sinks
>
> My code is here: https://pastebin.com/yjYCXAAR
>
> Thanks!