You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@beam.apache.org by Raghu Angadi <ra...@google.com.INVALID> on 2016/11/11 02:22:13 UTC
[PROPOSAL] Change to KafkaIO splits
I would like to propose a change to how many splits (sources) KafkaIO
creates. The code changes are relatively simple, but it has a couple of
drawbacks I would to discuss here.
KafkaIO currently takes '*desiredNumWorkers
<https://github.com/apache/incubator-beam/blob/v0.3.0-incubating-RC1/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L642>*'
hint literally and returns exactly that many splits. If *desiredNumWorkers*
is 10, and the topic has 50 partitions, each Kafka source reads from 5
partitions.
The primary disadvantage is that runner dependent 'desiredNumWorkers' might
not be accurate. In Dataflow, it is particularly low when we set
'maxNumWorkers' (BEAM-958 <https://issues.apache.org/jira/browse/BEAM-958>).
In addition, number of partitions in Kafka is a really good indicator of
its parallelism.
I would like to change KafkaIO to return one split for each of the
partitions.
Pros:
- A partition is in fact the unit of parallelism in Kafka.
- Does not depend on 'desiredNumWorkers'.
- Little risk of having unreasonably large number of partitions (unlike
say a source with one split for file). Number of partitions tend to be on
the order of the Kafka cluster size.
Cons: mainly affects job update:
- Breaks updating existing job
<https://cloud.google.com/dataflow/pipelines/updating-a-pipeline> if it
is updated to newer version of KafkaIO. New version changes number of
splits returned, which is not allowed during update.
- I think this is a reasonable breakage at this stage.
- Vast majority of updates don't involve version change
- We could add a work around where user can explicitly set number of
splits in KafkaIO (this might be required to handle change in
partitions as
well, see below)
- Makes it a bit more difficult to support change in number of Kafka
partitions across an update.
- This is not a feature in KafkaIO yet. So not a new breakage.
- If we don't depend on 'desiredNumWorkers', there is no way for us
to know how many splits we had before the update. This is actually a
limitation of UnboundedSource API. UnboundedSource needs
multiple teaks to
support job update better. In that sense I don't think this should be a
blocker.
- A work around is to let user explicitly set number of splits. E.g.
- when a job starts, say we had 70 partitions and after some time
we add 10 more partitions.
- At runtime, each Kafka split notices these and can distribute
new partitions among existing 70.
- But when the job is updated, KafkaIO does not know that it had
only 70 partitions earlier.
- For this to work, user could set number of splits to 70
explicitly.
Thanks.
Raghu.
Re: [PROPOSAL] Change to KafkaIO splits
Posted by Raghu Angadi <ra...@google.com.INVALID>.
I agree with all of this, except
I think this also avoids the need to "remember" the original number of
> parallelism.
KafkaIO still need to decide how many splits it needs to return in
generateInitialSplits().
'Update' could be Dataflow specific concern. We could drop it for this
thread, though I think IO and runners do interact with each other and thus
should influence the api. But that might be slated for future.
On Fri, Nov 11, 2016 at 6:09 AM, Amit Sela <am...@gmail.com> wrote:
> +1
> I think this makes more sense then the existing form of a split that is
> made of several Kafka partitions since, as mentioned, Kafka partitions are
> in fact it's parallelism.
>
> As for supporting a change in the number of partitions (mainly, added
> partitions), I'll suggest something I brought up before, and might make
> more sense now:
> Hashing an UnboundedSource according to it's split's properties
> (topic-partition in this case). This will allow to key the stream by the
> source in a way that the reader's CheckpointMark is tied to the split, and
> if a "new split" is created (a new partition added to a topic the pipeline
> consumes) it's reader's state is non-existing (starting from
> latest/earlies), while the rest (of the readers) will pick-up where they
> left.
> I think this also avoids the need to "remember" the original number of
> parallelism.
>
> Thanks,
> Amit
>
> On Fri, Nov 11, 2016 at 4:22 AM Raghu Angadi <ra...@google.com.invalid>
> wrote:
>
> > I would like to propose a change to how many splits (sources) KafkaIO
> > creates. The code changes are relatively simple, but it has a couple of
> > drawbacks I would to discuss here.
> >
> > KafkaIO currently takes '*desiredNumWorkers
> > <
> > https://github.com/apache/incubator-beam/blob/v0.3.0-
> incubating-RC1/sdks/java/io/kafka/src/main/java/org/
> apache/beam/sdk/io/kafka/KafkaIO.java#L642
> > >*'
> > hint literally and returns exactly that many splits. If
> *desiredNumWorkers*
> > is 10, and the topic has 50 partitions, each Kafka source reads from 5
> > partitions.
> >
> > The primary disadvantage is that runner dependent 'desiredNumWorkers'
> might
> > not be accurate. In Dataflow, it is particularly low when we set
> > 'maxNumWorkers' (BEAM-958 <https://issues.apache.org/
> jira/browse/BEAM-958
> > >).
> > In addition, number of partitions in Kafka is a really good indicator of
> > its parallelism.
> >
> > I would like to change KafkaIO to return one split for each of the
> > partitions.
> >
> > Pros:
> >
> > - A partition is in fact the unit of parallelism in Kafka.
> > - Does not depend on 'desiredNumWorkers'.
> > - Little risk of having unreasonably large number of partitions
> (unlike
> > say a source with one split for file). Number of partitions tend to be
> > on
> > the order of the Kafka cluster size.
> >
> > Cons: mainly affects job update:
> >
> > - Breaks updating existing job
> > <https://cloud.google.com/dataflow/pipelines/updating-a-pipeline> if
> it
> > is updated to newer version of KafkaIO. New version changes number of
> > splits returned, which is not allowed during update.
> > - I think this is a reasonable breakage at this stage.
> > - Vast majority of updates don't involve version change
> > - We could add a work around where user can explicitly set number
> of
> > splits in KafkaIO (this might be required to handle change in
> > partitions as
> > well, see below)
> > - Makes it a bit more difficult to support change in number of Kafka
> > partitions across an update.
> > - This is not a feature in KafkaIO yet. So not a new breakage.
> > - If we don't depend on 'desiredNumWorkers', there is no way for us
> > to know how many splits we had before the update. This is actually
> a
> > limitation of UnboundedSource API. UnboundedSource needs
> > multiple teaks to
> > support job update better. In that sense I don't think this should
> > be a
> > blocker.
> > - A work around is to let user explicitly set number of splits.
> E.g.
> > - when a job starts, say we had 70 partitions and after some
> time
> > we add 10 more partitions.
> > - At runtime, each Kafka split notices these and can distribute
> > new partitions among existing 70.
> > - But when the job is updated, KafkaIO does not know that it had
> > only 70 partitions earlier.
> > - For this to work, user could set number of splits to 70
> > explicitly.
> >
> >
> > Thanks.
> > Raghu.
> >
>
Re: [PROPOSAL] Change to KafkaIO splits
Posted by Raghu Angadi <ra...@google.com.INVALID>.
On Sun, Nov 13, 2016 at 10:14 PM, Davor Bonaci <da...@google.com.invalid>
wrote:
> Luke is bringing up great questions, I think.
>
Yes, better handling of 'desiredNumSplits' by a runner would be very
useful. I wanted to limit my proposal to what a source like KafkaIO could
do on its own.
> My first impression is that the current state is "possibly under-split",
> and the proposal is to move us to "possibly over-split" state. Neither is
> the ideal solution,
>
I think of it more like : shift from "strictly obey parallelism suggested
by the runner" to "parallelism inherent to a Kafka stream".
Raghu.
Re: [PROPOSAL] Change to KafkaIO splits
Posted by Amit Sela <am...@gmail.com>.
For Kafka, I don't think you're over-splitting if you split according to
Kafka partitions.
If your backend provides enough parallelism, you'll get a 1-1 (Source
splits-to-Kafka partitions) parallelism from the KafkaIO today.
The problem is with the backend not providing enough parallelism:
- Current implementation: a split is assigned with more than 1 Kafka
partition to read from, and since a split-read-task (how's it called ?) is
executed by a single thread (just made an assumption which might not be
true to all runners..), you'll "round-robin" on assigned Kafka splits.
- "Over splitting": Each split-read-task is reading of exactly 1 Kafka
partition, but there are queued split-read-tasks.
Does this make sense ?
Looking at it this way, which is better is probably runner-dependent.. in
terms of performance, I don't think it would matter for Spark but the
over-splitting shouldn't degrade performance and it can handle dynamically
growing Kafka topics.
On Mon, Nov 14, 2016 at 8:14 AM Davor Bonaci <da...@google.com.invalid>
wrote:
> Luke is bringing up great questions, I think.
>
> My first impression is that the current state is "possibly under-split",
> and the proposal is to move us to "possibly over-split" state. Neither is
> the ideal solution, as I'm sure we can find scenarios when either is not
> performing well. That said, if we aren't really solving the problem (at
> this time), I can believe that "over-split" is better than "under-split".
>
> (Job update is not a Beam consideration at this time, so none of that
> applies.)
>
> Davor
>
> On Fri, Nov 11, 2016 at 7:55 AM, Lukasz Cwik <lc...@google.com.invalid>
> wrote:
>
> > Why is it that we don't generate initial splits after the pipeline has
> been
> > created and the runner is processing it?
> >
> > This would allow a runner to look at the old state of the pipeline and
> see
> > how many splits there were.
> > This would allow the runner to provide a hint as to how many splits it
> > wants.
> > This brings it inline with how bounded sources work where the splitting
> is
> > performed once the pipeline has started.
> >
> > On Fri, Nov 11, 2016 at 8:09 AM, Amit Sela <am...@gmail.com> wrote:
> >
> > > +1
> > > I think this makes more sense then the existing form of a split that is
> > > made of several Kafka partitions since, as mentioned, Kafka partitions
> > are
> > > in fact it's parallelism.
> > >
> > > As for supporting a change in the number of partitions (mainly, added
> > > partitions), I'll suggest something I brought up before, and might make
> > > more sense now:
> > > Hashing an UnboundedSource according to it's split's properties
> > > (topic-partition in this case). This will allow to key the stream by
> the
> > > source in a way that the reader's CheckpointMark is tied to the split,
> > and
> > > if a "new split" is created (a new partition added to a topic the
> > pipeline
> > > consumes) it's reader's state is non-existing (starting from
> > > latest/earlies), while the rest (of the readers) will pick-up where
> they
> > > left.
> > > I think this also avoids the need to "remember" the original number of
> > > parallelism.
> > >
> > > Thanks,
> > > Amit
> > >
> > > On Fri, Nov 11, 2016 at 4:22 AM Raghu Angadi
> <rangadi@google.com.invalid
> > >
> > > wrote:
> > >
> > > > I would like to propose a change to how many splits (sources) KafkaIO
> > > > creates. The code changes are relatively simple, but it has a couple
> of
> > > > drawbacks I would to discuss here.
> > > >
> > > > KafkaIO currently takes '*desiredNumWorkers
> > > > <
> > > > https://github.com/apache/incubator-beam/blob/v0.3.0-
> > > incubating-RC1/sdks/java/io/kafka/src/main/java/org/
> > > apache/beam/sdk/io/kafka/KafkaIO.java#L642
> > > > >*'
> > > > hint literally and returns exactly that many splits. If
> > > *desiredNumWorkers*
> > > > is 10, and the topic has 50 partitions, each Kafka source reads from
> 5
> > > > partitions.
> > > >
> > > > The primary disadvantage is that runner dependent 'desiredNumWorkers'
> > > might
> > > > not be accurate. In Dataflow, it is particularly low when we set
> > > > 'maxNumWorkers' (BEAM-958 <https://issues.apache.org/
> > > jira/browse/BEAM-958
> > > > >).
> > > > In addition, number of partitions in Kafka is a really good indicator
> > of
> > > > its parallelism.
> > > >
> > > > I would like to change KafkaIO to return one split for each of the
> > > > partitions.
> > > >
> > > > Pros:
> > > >
> > > > - A partition is in fact the unit of parallelism in Kafka.
> > > > - Does not depend on 'desiredNumWorkers'.
> > > > - Little risk of having unreasonably large number of partitions
> > > (unlike
> > > > say a source with one split for file). Number of partitions tend
> to
> > be
> > > > on
> > > > the order of the Kafka cluster size.
> > > >
> > > > Cons: mainly affects job update:
> > > >
> > > > - Breaks updating existing job
> > > > <https://cloud.google.com/dataflow/pipelines/updating-a-pipeline>
> > if
> > > it
> > > > is updated to newer version of KafkaIO. New version changes number
> > of
> > > > splits returned, which is not allowed during update.
> > > > - I think this is a reasonable breakage at this stage.
> > > > - Vast majority of updates don't involve version change
> > > > - We could add a work around where user can explicitly set
> number
> > > of
> > > > splits in KafkaIO (this might be required to handle change in
> > > > partitions as
> > > > well, see below)
> > > > - Makes it a bit more difficult to support change in number of
> Kafka
> > > > partitions across an update.
> > > > - This is not a feature in KafkaIO yet. So not a new breakage.
> > > > - If we don't depend on 'desiredNumWorkers', there is no way
> for
> > us
> > > > to know how many splits we had before the update. This is
> > actually
> > > a
> > > > limitation of UnboundedSource API. UnboundedSource needs
> > > > multiple teaks to
> > > > support job update better. In that sense I don't think this
> > should
> > > > be a
> > > > blocker.
> > > > - A work around is to let user explicitly set number of splits.
> > > E.g.
> > > > - when a job starts, say we had 70 partitions and after some
> > > time
> > > > we add 10 more partitions.
> > > > - At runtime, each Kafka split notices these and can
> > distribute
> > > > new partitions among existing 70.
> > > > - But when the job is updated, KafkaIO does not know that it
> > had
> > > > only 70 partitions earlier.
> > > > - For this to work, user could set number of splits to 70
> > > > explicitly.
> > > >
> > > >
> > > > Thanks.
> > > > Raghu.
> > > >
> > >
> >
>
Re: [PROPOSAL] Change to KafkaIO splits
Posted by Davor Bonaci <da...@google.com.INVALID>.
Luke is bringing up great questions, I think.
My first impression is that the current state is "possibly under-split",
and the proposal is to move us to "possibly over-split" state. Neither is
the ideal solution, as I'm sure we can find scenarios when either is not
performing well. That said, if we aren't really solving the problem (at
this time), I can believe that "over-split" is better than "under-split".
(Job update is not a Beam consideration at this time, so none of that
applies.)
Davor
On Fri, Nov 11, 2016 at 7:55 AM, Lukasz Cwik <lc...@google.com.invalid>
wrote:
> Why is it that we don't generate initial splits after the pipeline has been
> created and the runner is processing it?
>
> This would allow a runner to look at the old state of the pipeline and see
> how many splits there were.
> This would allow the runner to provide a hint as to how many splits it
> wants.
> This brings it inline with how bounded sources work where the splitting is
> performed once the pipeline has started.
>
> On Fri, Nov 11, 2016 at 8:09 AM, Amit Sela <am...@gmail.com> wrote:
>
> > +1
> > I think this makes more sense then the existing form of a split that is
> > made of several Kafka partitions since, as mentioned, Kafka partitions
> are
> > in fact it's parallelism.
> >
> > As for supporting a change in the number of partitions (mainly, added
> > partitions), I'll suggest something I brought up before, and might make
> > more sense now:
> > Hashing an UnboundedSource according to it's split's properties
> > (topic-partition in this case). This will allow to key the stream by the
> > source in a way that the reader's CheckpointMark is tied to the split,
> and
> > if a "new split" is created (a new partition added to a topic the
> pipeline
> > consumes) it's reader's state is non-existing (starting from
> > latest/earlies), while the rest (of the readers) will pick-up where they
> > left.
> > I think this also avoids the need to "remember" the original number of
> > parallelism.
> >
> > Thanks,
> > Amit
> >
> > On Fri, Nov 11, 2016 at 4:22 AM Raghu Angadi <rangadi@google.com.invalid
> >
> > wrote:
> >
> > > I would like to propose a change to how many splits (sources) KafkaIO
> > > creates. The code changes are relatively simple, but it has a couple of
> > > drawbacks I would to discuss here.
> > >
> > > KafkaIO currently takes '*desiredNumWorkers
> > > <
> > > https://github.com/apache/incubator-beam/blob/v0.3.0-
> > incubating-RC1/sdks/java/io/kafka/src/main/java/org/
> > apache/beam/sdk/io/kafka/KafkaIO.java#L642
> > > >*'
> > > hint literally and returns exactly that many splits. If
> > *desiredNumWorkers*
> > > is 10, and the topic has 50 partitions, each Kafka source reads from 5
> > > partitions.
> > >
> > > The primary disadvantage is that runner dependent 'desiredNumWorkers'
> > might
> > > not be accurate. In Dataflow, it is particularly low when we set
> > > 'maxNumWorkers' (BEAM-958 <https://issues.apache.org/
> > jira/browse/BEAM-958
> > > >).
> > > In addition, number of partitions in Kafka is a really good indicator
> of
> > > its parallelism.
> > >
> > > I would like to change KafkaIO to return one split for each of the
> > > partitions.
> > >
> > > Pros:
> > >
> > > - A partition is in fact the unit of parallelism in Kafka.
> > > - Does not depend on 'desiredNumWorkers'.
> > > - Little risk of having unreasonably large number of partitions
> > (unlike
> > > say a source with one split for file). Number of partitions tend to
> be
> > > on
> > > the order of the Kafka cluster size.
> > >
> > > Cons: mainly affects job update:
> > >
> > > - Breaks updating existing job
> > > <https://cloud.google.com/dataflow/pipelines/updating-a-pipeline>
> if
> > it
> > > is updated to newer version of KafkaIO. New version changes number
> of
> > > splits returned, which is not allowed during update.
> > > - I think this is a reasonable breakage at this stage.
> > > - Vast majority of updates don't involve version change
> > > - We could add a work around where user can explicitly set number
> > of
> > > splits in KafkaIO (this might be required to handle change in
> > > partitions as
> > > well, see below)
> > > - Makes it a bit more difficult to support change in number of Kafka
> > > partitions across an update.
> > > - This is not a feature in KafkaIO yet. So not a new breakage.
> > > - If we don't depend on 'desiredNumWorkers', there is no way for
> us
> > > to know how many splits we had before the update. This is
> actually
> > a
> > > limitation of UnboundedSource API. UnboundedSource needs
> > > multiple teaks to
> > > support job update better. In that sense I don't think this
> should
> > > be a
> > > blocker.
> > > - A work around is to let user explicitly set number of splits.
> > E.g.
> > > - when a job starts, say we had 70 partitions and after some
> > time
> > > we add 10 more partitions.
> > > - At runtime, each Kafka split notices these and can
> distribute
> > > new partitions among existing 70.
> > > - But when the job is updated, KafkaIO does not know that it
> had
> > > only 70 partitions earlier.
> > > - For this to work, user could set number of splits to 70
> > > explicitly.
> > >
> > >
> > > Thanks.
> > > Raghu.
> > >
> >
>
Re: [PROPOSAL] Change to KafkaIO splits
Posted by Lukasz Cwik <lc...@google.com.INVALID>.
Why is it that we don't generate initial splits after the pipeline has been
created and the runner is processing it?
This would allow a runner to look at the old state of the pipeline and see
how many splits there were.
This would allow the runner to provide a hint as to how many splits it
wants.
This brings it inline with how bounded sources work where the splitting is
performed once the pipeline has started.
On Fri, Nov 11, 2016 at 8:09 AM, Amit Sela <am...@gmail.com> wrote:
> +1
> I think this makes more sense then the existing form of a split that is
> made of several Kafka partitions since, as mentioned, Kafka partitions are
> in fact it's parallelism.
>
> As for supporting a change in the number of partitions (mainly, added
> partitions), I'll suggest something I brought up before, and might make
> more sense now:
> Hashing an UnboundedSource according to it's split's properties
> (topic-partition in this case). This will allow to key the stream by the
> source in a way that the reader's CheckpointMark is tied to the split, and
> if a "new split" is created (a new partition added to a topic the pipeline
> consumes) it's reader's state is non-existing (starting from
> latest/earlies), while the rest (of the readers) will pick-up where they
> left.
> I think this also avoids the need to "remember" the original number of
> parallelism.
>
> Thanks,
> Amit
>
> On Fri, Nov 11, 2016 at 4:22 AM Raghu Angadi <ra...@google.com.invalid>
> wrote:
>
> > I would like to propose a change to how many splits (sources) KafkaIO
> > creates. The code changes are relatively simple, but it has a couple of
> > drawbacks I would to discuss here.
> >
> > KafkaIO currently takes '*desiredNumWorkers
> > <
> > https://github.com/apache/incubator-beam/blob/v0.3.0-
> incubating-RC1/sdks/java/io/kafka/src/main/java/org/
> apache/beam/sdk/io/kafka/KafkaIO.java#L642
> > >*'
> > hint literally and returns exactly that many splits. If
> *desiredNumWorkers*
> > is 10, and the topic has 50 partitions, each Kafka source reads from 5
> > partitions.
> >
> > The primary disadvantage is that runner dependent 'desiredNumWorkers'
> might
> > not be accurate. In Dataflow, it is particularly low when we set
> > 'maxNumWorkers' (BEAM-958 <https://issues.apache.org/
> jira/browse/BEAM-958
> > >).
> > In addition, number of partitions in Kafka is a really good indicator of
> > its parallelism.
> >
> > I would like to change KafkaIO to return one split for each of the
> > partitions.
> >
> > Pros:
> >
> > - A partition is in fact the unit of parallelism in Kafka.
> > - Does not depend on 'desiredNumWorkers'.
> > - Little risk of having unreasonably large number of partitions
> (unlike
> > say a source with one split for file). Number of partitions tend to be
> > on
> > the order of the Kafka cluster size.
> >
> > Cons: mainly affects job update:
> >
> > - Breaks updating existing job
> > <https://cloud.google.com/dataflow/pipelines/updating-a-pipeline> if
> it
> > is updated to newer version of KafkaIO. New version changes number of
> > splits returned, which is not allowed during update.
> > - I think this is a reasonable breakage at this stage.
> > - Vast majority of updates don't involve version change
> > - We could add a work around where user can explicitly set number
> of
> > splits in KafkaIO (this might be required to handle change in
> > partitions as
> > well, see below)
> > - Makes it a bit more difficult to support change in number of Kafka
> > partitions across an update.
> > - This is not a feature in KafkaIO yet. So not a new breakage.
> > - If we don't depend on 'desiredNumWorkers', there is no way for us
> > to know how many splits we had before the update. This is actually
> a
> > limitation of UnboundedSource API. UnboundedSource needs
> > multiple teaks to
> > support job update better. In that sense I don't think this should
> > be a
> > blocker.
> > - A work around is to let user explicitly set number of splits.
> E.g.
> > - when a job starts, say we had 70 partitions and after some
> time
> > we add 10 more partitions.
> > - At runtime, each Kafka split notices these and can distribute
> > new partitions among existing 70.
> > - But when the job is updated, KafkaIO does not know that it had
> > only 70 partitions earlier.
> > - For this to work, user could set number of splits to 70
> > explicitly.
> >
> >
> > Thanks.
> > Raghu.
> >
>
Re: [PROPOSAL] Change to KafkaIO splits
Posted by Amit Sela <am...@gmail.com>.
+1
I think this makes more sense then the existing form of a split that is
made of several Kafka partitions since, as mentioned, Kafka partitions are
in fact it's parallelism.
As for supporting a change in the number of partitions (mainly, added
partitions), I'll suggest something I brought up before, and might make
more sense now:
Hashing an UnboundedSource according to it's split's properties
(topic-partition in this case). This will allow to key the stream by the
source in a way that the reader's CheckpointMark is tied to the split, and
if a "new split" is created (a new partition added to a topic the pipeline
consumes) it's reader's state is non-existing (starting from
latest/earlies), while the rest (of the readers) will pick-up where they
left.
I think this also avoids the need to "remember" the original number of
parallelism.
Thanks,
Amit
On Fri, Nov 11, 2016 at 4:22 AM Raghu Angadi <ra...@google.com.invalid>
wrote:
> I would like to propose a change to how many splits (sources) KafkaIO
> creates. The code changes are relatively simple, but it has a couple of
> drawbacks I would to discuss here.
>
> KafkaIO currently takes '*desiredNumWorkers
> <
> https://github.com/apache/incubator-beam/blob/v0.3.0-incubating-RC1/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L642
> >*'
> hint literally and returns exactly that many splits. If *desiredNumWorkers*
> is 10, and the topic has 50 partitions, each Kafka source reads from 5
> partitions.
>
> The primary disadvantage is that runner dependent 'desiredNumWorkers' might
> not be accurate. In Dataflow, it is particularly low when we set
> 'maxNumWorkers' (BEAM-958 <https://issues.apache.org/jira/browse/BEAM-958
> >).
> In addition, number of partitions in Kafka is a really good indicator of
> its parallelism.
>
> I would like to change KafkaIO to return one split for each of the
> partitions.
>
> Pros:
>
> - A partition is in fact the unit of parallelism in Kafka.
> - Does not depend on 'desiredNumWorkers'.
> - Little risk of having unreasonably large number of partitions (unlike
> say a source with one split for file). Number of partitions tend to be
> on
> the order of the Kafka cluster size.
>
> Cons: mainly affects job update:
>
> - Breaks updating existing job
> <https://cloud.google.com/dataflow/pipelines/updating-a-pipeline> if it
> is updated to newer version of KafkaIO. New version changes number of
> splits returned, which is not allowed during update.
> - I think this is a reasonable breakage at this stage.
> - Vast majority of updates don't involve version change
> - We could add a work around where user can explicitly set number of
> splits in KafkaIO (this might be required to handle change in
> partitions as
> well, see below)
> - Makes it a bit more difficult to support change in number of Kafka
> partitions across an update.
> - This is not a feature in KafkaIO yet. So not a new breakage.
> - If we don't depend on 'desiredNumWorkers', there is no way for us
> to know how many splits we had before the update. This is actually a
> limitation of UnboundedSource API. UnboundedSource needs
> multiple teaks to
> support job update better. In that sense I don't think this should
> be a
> blocker.
> - A work around is to let user explicitly set number of splits. E.g.
> - when a job starts, say we had 70 partitions and after some time
> we add 10 more partitions.
> - At runtime, each Kafka split notices these and can distribute
> new partitions among existing 70.
> - But when the job is updated, KafkaIO does not know that it had
> only 70 partitions earlier.
> - For this to work, user could set number of splits to 70
> explicitly.
>
>
> Thanks.
> Raghu.
>