[jira] [Commented] (KAFKA-4601) Avoid duplicated repartitioning in KStream DSL
[
https://issues.apache.org/jira/browse/KAFKA-4601?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15894770#comment-15894770
]
Matthias J. Sax commented on KAFKA-4601:
Thanks!
> Avoid duplicated repartitioning in KStream DSL
> --
>
> Key: KAFKA-4601
> URL: https://issues.apache.org/jira/browse/KAFKA-4601
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
> Labels: performance
>
> Consider the following DSL:
> {code}
> Stream source = builder.stream(Serdes.String(),
> Serdes.String(), "topic1").map(..);
> KTable counts = source
> .groupByKey()
> .count("Counts");
> KStream sink = source.leftJoin(counts, ..);
> {code}
> The resulted topology looks like this:
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-FILTER-04, KSTREAM-FILTER-07]
> KSTREAM-FILTER-04:
> children:
> [KSTREAM-SINK-03]
> KSTREAM-SINK-03:
> topic: X-Counts-repartition
> KSTREAM-FILTER-07:
> children:
> [KSTREAM-SINK-06]
> KSTREAM-SINK-06:
> topic:
> X-KSTREAM-MAP-01-repartition
> ProcessorTopology:
> KSTREAM-SOURCE-08:
> topics:
> [X-KSTREAM-MAP-01-repartition]
> children:
> [KSTREAM-LEFTJOIN-09]
> KSTREAM-LEFTJOIN-09:
> states: [Counts]
> KSTREAM-SOURCE-05:
> topics: [X-Counts-repartition]
> children:
> [KSTREAM-AGGREGATE-02]
> KSTREAM-AGGREGATE-02:
> states: [Counts]
> {code}
> I.e. there are two repartition topics, one for the aggregate and one for the
> join, which not only introduce unnecessary overheads but also mess up the
> processing ordering (users are expecting each record to go through
> aggregation first then the join operator). And in order to get the following
> simpler topology users today need to add a {{through}} operator after {{map}}
> manually to enforce repartitioning.
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-SINK-02]
> KSTREAM-SINK-02:
> topic: topic 2
> ProcessorTopology:
> KSTREAM-SOURCE-03:
> topics: [topic 2]
> children:
> [KSTREAM-AGGREGATE-04, KSTREAM-LEFTJOIN-05]
> KSTREAM-AGGREGATE-04:
> states: [Counts]
> KSTREAM-LEFTJOIN-05:
> states: [Counts]
> {code}
> This kind of optimization should be automatic in Streams, which we can
> consider doing when extending from one-operator-at-a-time translation.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
[jira] [Commented] (KAFKA-4601) Avoid duplicated repartitioning in KStream DSL
[
https://issues.apache.org/jira/browse/KAFKA-4601?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15894498#comment-15894498
]
Michal Borowiecki commented on KAFKA-4601:
--
Created KAFKA-4835.
> Avoid duplicated repartitioning in KStream DSL
> --
>
> Key: KAFKA-4601
> URL: https://issues.apache.org/jira/browse/KAFKA-4601
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
> Labels: performance
>
> Consider the following DSL:
> {code}
> Stream source = builder.stream(Serdes.String(),
> Serdes.String(), "topic1").map(..);
> KTable counts = source
> .groupByKey()
> .count("Counts");
> KStream sink = source.leftJoin(counts, ..);
> {code}
> The resulted topology looks like this:
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-FILTER-04, KSTREAM-FILTER-07]
> KSTREAM-FILTER-04:
> children:
> [KSTREAM-SINK-03]
> KSTREAM-SINK-03:
> topic: X-Counts-repartition
> KSTREAM-FILTER-07:
> children:
> [KSTREAM-SINK-06]
> KSTREAM-SINK-06:
> topic:
> X-KSTREAM-MAP-01-repartition
> ProcessorTopology:
> KSTREAM-SOURCE-08:
> topics:
> [X-KSTREAM-MAP-01-repartition]
> children:
> [KSTREAM-LEFTJOIN-09]
> KSTREAM-LEFTJOIN-09:
> states: [Counts]
> KSTREAM-SOURCE-05:
> topics: [X-Counts-repartition]
> children:
> [KSTREAM-AGGREGATE-02]
> KSTREAM-AGGREGATE-02:
> states: [Counts]
> {code}
> I.e. there are two repartition topics, one for the aggregate and one for the
> join, which not only introduce unnecessary overheads but also mess up the
> processing ordering (users are expecting each record to go through
> aggregation first then the join operator). And in order to get the following
> simpler topology users today need to add a {{through}} operator after {{map}}
> manually to enforce repartitioning.
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-SINK-02]
> KSTREAM-SINK-02:
> topic: topic 2
> ProcessorTopology:
> KSTREAM-SOURCE-03:
> topics: [topic 2]
> children:
> [KSTREAM-AGGREGATE-04, KSTREAM-LEFTJOIN-05]
> KSTREAM-AGGREGATE-04:
> states: [Counts]
> KSTREAM-LEFTJOIN-05:
> states: [Counts]
> {code}
> This kind of optimization should be automatic in Streams, which we can
> consider doing when extending from one-operator-at-a-time translation.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
[jira] [Commented] (KAFKA-4601) Avoid duplicated repartitioning in KStream DSL
[
https://issues.apache.org/jira/browse/KAFKA-4601?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15881339#comment-15881339
]
Matthias J. Sax commented on KAFKA-4601:
[~mihbor] Thanks for your comment. I guess it's related, but IMHO should be an
own JIRA. Would you mind creating one? I understand your need and had a similar
though like this once too. But it's a "dangerous" feature, too. But we discuss
in the new JIRA about pros/cons and if we should allow this or not. Thanks.
> Avoid duplicated repartitioning in KStream DSL
> --
>
> Key: KAFKA-4601
> URL: https://issues.apache.org/jira/browse/KAFKA-4601
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
> Labels: performance
>
> Consider the following DSL:
> {code}
> Stream source = builder.stream(Serdes.String(),
> Serdes.String(), "topic1").map(..);
> KTable counts = source
> .groupByKey()
> .count("Counts");
> KStream sink = source.leftJoin(counts, ..);
> {code}
> The resulted topology looks like this:
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-FILTER-04, KSTREAM-FILTER-07]
> KSTREAM-FILTER-04:
> children:
> [KSTREAM-SINK-03]
> KSTREAM-SINK-03:
> topic: X-Counts-repartition
> KSTREAM-FILTER-07:
> children:
> [KSTREAM-SINK-06]
> KSTREAM-SINK-06:
> topic:
> X-KSTREAM-MAP-01-repartition
> ProcessorTopology:
> KSTREAM-SOURCE-08:
> topics:
> [X-KSTREAM-MAP-01-repartition]
> children:
> [KSTREAM-LEFTJOIN-09]
> KSTREAM-LEFTJOIN-09:
> states: [Counts]
> KSTREAM-SOURCE-05:
> topics: [X-Counts-repartition]
> children:
> [KSTREAM-AGGREGATE-02]
> KSTREAM-AGGREGATE-02:
> states: [Counts]
> {code}
> I.e. there are two repartition topics, one for the aggregate and one for the
> join, which not only introduce unnecessary overheads but also mess up the
> processing ordering (users are expecting each record to go through
> aggregation first then the join operator). And in order to get the following
> simpler topology users today need to add a {{through}} operator after {{map}}
> manually to enforce repartitioning.
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-SINK-02]
> KSTREAM-SINK-02:
> topic: topic 2
> ProcessorTopology:
> KSTREAM-SOURCE-03:
> topics: [topic 2]
> children:
> [KSTREAM-AGGREGATE-04, KSTREAM-LEFTJOIN-05]
> KSTREAM-AGGREGATE-04:
> states: [Counts]
> KSTREAM-LEFTJOIN-05:
> states: [Counts]
> {code}
> This kind of optimization should be automatic in Streams, which we can
> consider doing when extending from one-operator-at-a-time translation.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
[jira] [Commented] (KAFKA-4601) Avoid duplicated repartitioning in KStream DSL
[
https://issues.apache.org/jira/browse/KAFKA-4601?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15881030#comment-15881030
]
Michal Borowiecki commented on KAFKA-4601:
--
Don't know if this belongs in this ticket or warrants a separate one, but I'd
suggest, instead of trying to rely on kstreams doing more automatic
optimization, it would be good to provide users more control over the
repartitioning.
My use case is as follows (unrelated bits omitted for brevity):
{code}
KTable loggedInCustomers = builder
.stream("customerLogins")
.groupBy((key, activity) ->
activity.getCustomerRef())
.reduce((first,second) -> second, loginStore());
builder
.stream("balanceUpdates")
.map((key, activity) -> new KeyValue<>(
activity.getCustomerRef(),
activity))
.join(loggedInCustomers, (activity, session) -> ...
.to("sessions");
{code}
Both "groupBy" and "map" in the underlying implementation set the
repartitionRequired flag (since the key changes), and the aggregation/join that
follows will create the repartitioned topic.
However, in our case I know that both input streams are already partitioned by
the customerRef value, which I'm mapping into the key (because it's required by
the join operation).
So there are 2 unnecessary intermediate topics created with their associated
overhead, while the ultimate goal is simply to do a join on a value that we
already use to partition the original streams anyway.
(Note, we don't have the option to re-implement the original input streams to
make customerRef the message key.)
I think it would be better to allow the user to decide (from their knowledge of
the incoming streams) whether a repartition is mandatory on aggregation and
join operations (overloaded version of the methods with the repartitionRequired
flag exposed maybe?)
An alternative would be to allow users to perform a join on a value other than
the key (a keyValueMapper parameter to join, like the one used for joins with
global tables), but I expect that to be more involved and error-prone to use
for people who don't understand the partitioning requirements well (whereas
it's safe for global tables).
> Avoid duplicated repartitioning in KStream DSL
> --
>
> Key: KAFKA-4601
> URL: https://issues.apache.org/jira/browse/KAFKA-4601
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
> Labels: performance
>
> Consider the following DSL:
> {code}
> Stream source = builder.stream(Serdes.String(),
> Serdes.String(), "topic1").map(..);
> KTable counts = source
> .groupByKey()
> .count("Counts");
> KStream sink = source.leftJoin(counts, ..);
> {code}
> The resulted topology looks like this:
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-FILTER-04, KSTREAM-FILTER-07]
> KSTREAM-FILTER-04:
> children:
> [KSTREAM-SINK-03]
> KSTREAM-SINK-03:
> topic: X-Counts-repartition
> KSTREAM-FILTER-07:
> children:
> [KSTREAM-SINK-06]
> KSTREAM-SINK-06:
> topic:
> X-KSTREAM-MAP-01-repartition
> ProcessorTopology:
> KSTREAM-SOURCE-08:
> topics:
> [X-KSTREAM-MAP-01-repartition]
> children:
> [KSTREAM-LEFTJOIN-09]
> KSTREAM-LEFTJOIN-09:
> states: [Counts]
> KSTREAM-SOURCE-05:
> topics: [X-Counts-repartition]
> children:
> [KSTREAM-AGGREGATE-02]
> KSTREAM-AGGREGATE-02:
> states:
[jira] [Commented] (KAFKA-4601) Avoid duplicated repartitioning in KStream DSL
[
https://issues.apache.org/jira/browse/KAFKA-4601?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15802932#comment-15802932
]
Bill Bejeck commented on KAFKA-4601:
Thanks for the explanation. Avoiding duplicated partitioning sounds more
involved than I may have time to devote to it.
> Avoid duplicated repartitioning in KStream DSL
> --
>
> Key: KAFKA-4601
> URL: https://issues.apache.org/jira/browse/KAFKA-4601
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
>Assignee: Bill Bejeck
> Labels: performance
>
> Consider the following DSL:
> {code}
> Stream source = builder.stream(Serdes.String(),
> Serdes.String(), "topic1").map(..);
> KTable counts = source
> .groupByKey()
> .count("Counts");
> KStream sink = source.leftJoin(counts, ..);
> {code}
> The resulted topology looks like this:
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-FILTER-04, KSTREAM-FILTER-07]
> KSTREAM-FILTER-04:
> children:
> [KSTREAM-SINK-03]
> KSTREAM-SINK-03:
> topic: X-Counts-repartition
> KSTREAM-FILTER-07:
> children:
> [KSTREAM-SINK-06]
> KSTREAM-SINK-06:
> topic:
> X-KSTREAM-MAP-01-repartition
> ProcessorTopology:
> KSTREAM-SOURCE-08:
> topics:
> [X-KSTREAM-MAP-01-repartition]
> children:
> [KSTREAM-LEFTJOIN-09]
> KSTREAM-LEFTJOIN-09:
> states: [Counts]
> KSTREAM-SOURCE-05:
> topics: [X-Counts-repartition]
> children:
> [KSTREAM-AGGREGATE-02]
> KSTREAM-AGGREGATE-02:
> states: [Counts]
> {code}
> I.e. there are two repartition topics, one for the aggregate and one for the
> join, which not only introduce unnecessary overheads but also mess up the
> processing ordering (users are expecting each record to go through
> aggregation first then the join operator). And in order to get the following
> simpler topology users today need to add a {{through}} operator after {{map}}
> manually to enforce repartitioning.
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-SINK-02]
> KSTREAM-SINK-02:
> topic: topic 2
> ProcessorTopology:
> KSTREAM-SOURCE-03:
> topics: [topic 2]
> children:
> [KSTREAM-AGGREGATE-04, KSTREAM-LEFTJOIN-05]
> KSTREAM-AGGREGATE-04:
> states: [Counts]
> KSTREAM-LEFTJOIN-05:
> states: [Counts]
> {code}
> This kind of optimization should be automatic in Streams, which we can
> consider doing when extending from one-operator-at-a-time translation.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (KAFKA-4601) Avoid duplicated repartitioning in KStream DSL
[
https://issues.apache.org/jira/browse/KAFKA-4601?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15802904#comment-15802904
]
Guozhang Wang commented on KAFKA-4601:
--
[~bbejeck] Note that it may be more complex than it sounds: today we translate
the DSL operators to underlying processor nodes in topology independently (i.e.
one at a time), so when we are translating a join after an aggregate, we do not
know what processors have been created so far. To solve this specific issue we
can make some workaround, but a general solution would be extending the
translation mechanism to be "global": this can be as complex as query
optimization.
> Avoid duplicated repartitioning in KStream DSL
> --
>
> Key: KAFKA-4601
> URL: https://issues.apache.org/jira/browse/KAFKA-4601
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
>Assignee: Bill Bejeck
> Labels: performance
>
> Consider the following DSL:
> {code}
> Stream source = builder.stream(Serdes.String(),
> Serdes.String(), "topic1").map(..);
> KTable counts = source
> .groupByKey()
> .count("Counts");
> KStream sink = source.leftJoin(counts, ..);
> {code}
> The resulted topology looks like this:
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-FILTER-04, KSTREAM-FILTER-07]
> KSTREAM-FILTER-04:
> children:
> [KSTREAM-SINK-03]
> KSTREAM-SINK-03:
> topic: X-Counts-repartition
> KSTREAM-FILTER-07:
> children:
> [KSTREAM-SINK-06]
> KSTREAM-SINK-06:
> topic:
> X-KSTREAM-MAP-01-repartition
> ProcessorTopology:
> KSTREAM-SOURCE-08:
> topics:
> [X-KSTREAM-MAP-01-repartition]
> children:
> [KSTREAM-LEFTJOIN-09]
> KSTREAM-LEFTJOIN-09:
> states: [Counts]
> KSTREAM-SOURCE-05:
> topics: [X-Counts-repartition]
> children:
> [KSTREAM-AGGREGATE-02]
> KSTREAM-AGGREGATE-02:
> states: [Counts]
> {code}
> I.e. there are two repartition topics, one for the aggregate and one for the
> join, which not only introduce unnecessary overheads but also mess up the
> processing ordering (users are expecting each record to go through
> aggregation first then the join operator). And in order to get the following
> simpler topology users today need to add a {{through}} operator after {{map}}
> manually to enforce repartitioning.
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-SINK-02]
> KSTREAM-SINK-02:
> topic: topic 2
> ProcessorTopology:
> KSTREAM-SOURCE-03:
> topics: [topic 2]
> children:
> [KSTREAM-AGGREGATE-04, KSTREAM-LEFTJOIN-05]
> KSTREAM-AGGREGATE-04:
> states: [Counts]
> KSTREAM-LEFTJOIN-05:
> states: [Counts]
> {code}
> This kind of optimization should be automatic in Streams, which we can
> consider doing when extending from one-operator-at-a-time translation.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (KAFKA-4601) Avoid duplicated repartitioning in KStream DSL
[
https://issues.apache.org/jira/browse/KAFKA-4601?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15802878#comment-15802878
]
Bill Bejeck commented on KAFKA-4601:
If it's ok, picking this one up.
> Avoid duplicated repartitioning in KStream DSL
> --
>
> Key: KAFKA-4601
> URL: https://issues.apache.org/jira/browse/KAFKA-4601
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
>Assignee: Bill Bejeck
> Labels: performance
>
> Consider the following DSL:
> {code}
> Stream source = builder.stream(Serdes.String(),
> Serdes.String(), "topic1").map(..);
> KTable counts = source
> .groupByKey()
> .count("Counts");
> KStream sink = source.leftJoin(counts, ..);
> {code}
> The resulted topology looks like this:
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-FILTER-04, KSTREAM-FILTER-07]
> KSTREAM-FILTER-04:
> children:
> [KSTREAM-SINK-03]
> KSTREAM-SINK-03:
> topic: X-Counts-repartition
> KSTREAM-FILTER-07:
> children:
> [KSTREAM-SINK-06]
> KSTREAM-SINK-06:
> topic:
> X-KSTREAM-MAP-01-repartition
> ProcessorTopology:
> KSTREAM-SOURCE-08:
> topics:
> [X-KSTREAM-MAP-01-repartition]
> children:
> [KSTREAM-LEFTJOIN-09]
> KSTREAM-LEFTJOIN-09:
> states: [Counts]
> KSTREAM-SOURCE-05:
> topics: [X-Counts-repartition]
> children:
> [KSTREAM-AGGREGATE-02]
> KSTREAM-AGGREGATE-02:
> states: [Counts]
> {code}
> I.e. there are two repartition topics, one for the aggregate and one for the
> join, which not only introduce unnecessary overheads but also mess up the
> processing ordering (users are expecting each record to go through
> aggregation first then the join operator). And in order to get the following
> simpler topology users today need to add a {{through}} operator after {{map}}
> manually to enforce repartitioning.
> {code}
> ProcessorTopology:
> KSTREAM-SOURCE-00:
> topics: [topic1]
> children: [KSTREAM-MAP-01]
> KSTREAM-MAP-01:
> children:
> [KSTREAM-SINK-02]
> KSTREAM-SINK-02:
> topic: topic 2
> ProcessorTopology:
> KSTREAM-SOURCE-03:
> topics: [topic 2]
> children:
> [KSTREAM-AGGREGATE-04, KSTREAM-LEFTJOIN-05]
> KSTREAM-AGGREGATE-04:
> states: [Counts]
> KSTREAM-LEFTJOIN-05:
> states: [Counts]
> {code}
> This kind of optimization should be automatic in Streams, which we can
> consider doing when extending from one-operator-at-a-time translation.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
