You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Artur Mrozowski <ar...@gmail.com> on 2017/11/22 11:02:09 UTC
Joins in Kafka Streams and partitioning of the topics
Hi,
I am joining 4 different topic with 4 partitions each using 0.10.0.0
version of Kafka Streams. The joins are KTable to KTable. Is there
anything I should be aware of considering partitions or version of Kafka
Streams? In other words should I be expecting consistent results or do I
need to for example use Global tables.
I'd like to run that application on Kubernetes later on. Should I think of
anything or do different instances of the same Kafka Streams application
take care of management of the state?
Grateful for any thoughts or a piece of advice
Best Regards
/Artur
Re: Joins in Kafka Streams and partitioning of the topics
Posted by Artur Mrozowski <ar...@gmail.com>.
Yes you are probably right. So I was inspired be the KIP 150 blog post, so
the entire statement would be like this:
KTable<Integer,CustomerList> customerGrouped=
kStreamBuilder.stream(stringSerde, customerMessageSerde,
CUSTOMER_TOPIC)
.groupBy((key,value) ->
Integer.parseInt(value.customer.replaceFirst("cust","")),integerSerde,customerMessageSerde)
.aggregate(CustomerList::new,(ckey, custMessage,
customerList) -> {
customerList.lst.add(custMessage);
return customerList;
},customerListSerde,CUSTOMER_STORE);
The second oddity is beacause I want to gather history, all the previous
records. Exactly the same logic as the shopping cart , purchases and
wishList described in KIP 150. The result of the joins will contain all
the history for particular key.
You mention repartioning. I also have a feeling thea GlobaKTables are more
suitable for look ups and not what I am trying to do here.
How do I repartition the topics? Is it by using keyed messages or are there
other alternatives? I've noticed that when I run two instances of the
application than each have only half of the records posted to the 4
partitions topic.
I'd be happy to join the slack, do I need an invite?
/Artur
n
On Thu, Nov 30, 2017 at 9:30 PM, Jan Filipiak <Ja...@trivago.com>
wrote:
> There are some oddities in your topology that make make we wonder if they
> are the true drivers of your question.
>
> https://github.com/afuyo/KStreamsDemo/blob/master/src/main/
> java/kstream.demo/CustomerStreamPipelineHDI.java#L300
> Feels like it should be a KTable to begin with for example otherwise it is
> not clear how big this is supposed to grow
> https://github.com/afuyo/KStreamsDemo/blob/master/src/main/
> java/kstream.demo/CustomerStreamPipelineHDI.java#L325
> Same thing for policies. KGlobalTable might be chipped in later if you fat
> up from too many repartitions as some sort of
> performance optimisation, but my opinions on it are not to high.
>
>
> Hope that helps, just keep the questions coming, also check if you might
> want to join confluentcommunity on slack.
> Could never imaging that something like a insurance can really be modelled
> as 4 streams ;)
>
> Best Jan
>
>
>
>
>
>
> On 30.11.2017 21:07, Artur Mrozowski wrote:
>
>> what if I start two instances of that application? Does the state migrate
>> between the applications? Is it then I have to use a global table?
>>
>> BR
>> Artur
>>
>> On Thu, Nov 30, 2017 at 7:40 PM, Jan Filipiak <Ja...@trivago.com>
>> wrote:
>>
>> Hi,
>>>
>>> Haven't checked your code. But from what you describe you should be fine.
>>> Upgrading the version might help here and there but should still work
>>> with
>>> 0.10
>>> I guess.
>>>
>>> Best Jan
>>>
>>>
>>>
>>> On 30.11.2017 19:16, Artur Mrozowski wrote:
>>>
>>> Thank you Damian, it was very helpful.
>>>> I have implemented my solution in version 0.11.0.2 but there is one
>>>> thing
>>>> I
>>>> still wonder.
>>>> So what I try to do is what is described in KIP 150. Since it didn't
>>>> make
>>>> to the release for 1.0 I do it the old fashioned way.
>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-150+-+
>>>> Kafka-Streams+Cogroup
>>>> First
>>>> KTable<K, V1> table1 =
>>>> builder.stream("topic1").groupByKey().aggregate(initializer1,
>>>> aggregator1, aggValueSerde1, storeName1);
>>>>
>>>>
>>>> for all the four topics and then I join the results.
>>>> And here is the thing, the topics are partitioned and I don't used
>>>> global
>>>> tables, nor keyed messages and it seems to work fine.
>>>>
>>>> From Confluents documentation one could get impression that when
>>>> reading
>>>> from partitoned topics you need to use global tables. But is it really
>>>> necessary in this case?
>>>> And if not then why?
>>>>
>>>> Thanks again
>>>> Artur
>>>>
>>>> Here is the link to my implementation
>>>>
>>>> https://github.com/afuyo/KStreamsDemo/blob/master/src/main/
>>>> java/kstream.demo/CustomerStreamPipelineHDI.java
>>>>
>>>> On Wed, Nov 22, 2017 at 12:10 PM, Damian Guy <da...@gmail.com>
>>>> wrote:
>>>>
>>>> Hi Artur,
>>>>
>>>>> KafkaStreams 0.10.0.0 is quite old and a lot has changed and been fixed
>>>>> since then. If possible i'd recommend upgrading to at least 0.11.0.2 or
>>>>> 1.0.
>>>>> For joins you need to ensure that the topics have the same number of
>>>>> partitions (which they do) and that they are keyed the same.
>>>>>
>>>>> Thanks,
>>>>> Damian
>>>>>
>>>>> On Wed, 22 Nov 2017 at 11:02 Artur Mrozowski <ar...@gmail.com> wrote:
>>>>>
>>>>> Hi,
>>>>>
>>>>>> I am joining 4 different topic with 4 partitions each using 0.10.0.0
>>>>>> version of Kafka Streams. The joins are KTable to KTable. Is there
>>>>>> anything I should be aware of considering partitions or version of
>>>>>> Kafka
>>>>>> Streams? In other words should I be expecting consistent results or
>>>>>> do I
>>>>>> need to for example use Global tables.
>>>>>>
>>>>>> I'd like to run that application on Kubernetes later on. Should I
>>>>>> think
>>>>>>
>>>>>> of
>>>>>
>>>>> anything or do different instances of the same Kafka Streams
>>>>>> application
>>>>>> take care of management of the state?
>>>>>>
>>>>>> Grateful for any thoughts or a piece of advice
>>>>>>
>>>>>> Best Regards
>>>>>> /Artur
>>>>>>
>>>>>>
>>>>>>
>
Re: Joins in Kafka Streams and partitioning of the topics
Posted by Jan Filipiak <Ja...@trivago.com>.
There are some oddities in your topology that make make we wonder if
they are the true drivers of your question.
https://github.com/afuyo/KStreamsDemo/blob/master/src/main/java/kstream.demo/CustomerStreamPipelineHDI.java#L300
Feels like it should be a KTable to begin with for example otherwise it
is not clear how big this is supposed to grow
https://github.com/afuyo/KStreamsDemo/blob/master/src/main/java/kstream.demo/CustomerStreamPipelineHDI.java#L325
Same thing for policies. KGlobalTable might be chipped in later if you
fat up from too many repartitions as some sort of
performance optimisation, but my opinions on it are not to high.
Hope that helps, just keep the questions coming, also check if you might
want to join confluentcommunity on slack.
Could never imaging that something like a insurance can really be
modelled as 4 streams ;)
Best Jan
On 30.11.2017 21:07, Artur Mrozowski wrote:
> what if I start two instances of that application? Does the state migrate
> between the applications? Is it then I have to use a global table?
>
> BR
> Artur
>
> On Thu, Nov 30, 2017 at 7:40 PM, Jan Filipiak <Ja...@trivago.com>
> wrote:
>
>> Hi,
>>
>> Haven't checked your code. But from what you describe you should be fine.
>> Upgrading the version might help here and there but should still work with
>> 0.10
>> I guess.
>>
>> Best Jan
>>
>>
>>
>> On 30.11.2017 19:16, Artur Mrozowski wrote:
>>
>>> Thank you Damian, it was very helpful.
>>> I have implemented my solution in version 0.11.0.2 but there is one thing
>>> I
>>> still wonder.
>>> So what I try to do is what is described in KIP 150. Since it didn't make
>>> to the release for 1.0 I do it the old fashioned way.
>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-150+-+
>>> Kafka-Streams+Cogroup
>>> First
>>> KTable<K, V1> table1 =
>>> builder.stream("topic1").groupByKey().aggregate(initializer1,
>>> aggregator1, aggValueSerde1, storeName1);
>>>
>>>
>>> for all the four topics and then I join the results.
>>> And here is the thing, the topics are partitioned and I don't used global
>>> tables, nor keyed messages and it seems to work fine.
>>>
>>> From Confluents documentation one could get impression that when reading
>>> from partitoned topics you need to use global tables. But is it really
>>> necessary in this case?
>>> And if not then why?
>>>
>>> Thanks again
>>> Artur
>>>
>>> Here is the link to my implementation
>>>
>>> https://github.com/afuyo/KStreamsDemo/blob/master/src/main/
>>> java/kstream.demo/CustomerStreamPipelineHDI.java
>>>
>>> On Wed, Nov 22, 2017 at 12:10 PM, Damian Guy <da...@gmail.com>
>>> wrote:
>>>
>>> Hi Artur,
>>>> KafkaStreams 0.10.0.0 is quite old and a lot has changed and been fixed
>>>> since then. If possible i'd recommend upgrading to at least 0.11.0.2 or
>>>> 1.0.
>>>> For joins you need to ensure that the topics have the same number of
>>>> partitions (which they do) and that they are keyed the same.
>>>>
>>>> Thanks,
>>>> Damian
>>>>
>>>> On Wed, 22 Nov 2017 at 11:02 Artur Mrozowski <ar...@gmail.com> wrote:
>>>>
>>>> Hi,
>>>>> I am joining 4 different topic with 4 partitions each using 0.10.0.0
>>>>> version of Kafka Streams. The joins are KTable to KTable. Is there
>>>>> anything I should be aware of considering partitions or version of Kafka
>>>>> Streams? In other words should I be expecting consistent results or do I
>>>>> need to for example use Global tables.
>>>>>
>>>>> I'd like to run that application on Kubernetes later on. Should I think
>>>>>
>>>> of
>>>>
>>>>> anything or do different instances of the same Kafka Streams application
>>>>> take care of management of the state?
>>>>>
>>>>> Grateful for any thoughts or a piece of advice
>>>>>
>>>>> Best Regards
>>>>> /Artur
>>>>>
>>>>>
Re: Joins in Kafka Streams and partitioning of the topics
Posted by Artur Mrozowski <ar...@gmail.com>.
what if I start two instances of that application? Does the state migrate
between the applications? Is it then I have to use a global table?
BR
Artur
On Thu, Nov 30, 2017 at 7:40 PM, Jan Filipiak <Ja...@trivago.com>
wrote:
> Hi,
>
> Haven't checked your code. But from what you describe you should be fine.
> Upgrading the version might help here and there but should still work with
> 0.10
> I guess.
>
> Best Jan
>
>
>
> On 30.11.2017 19:16, Artur Mrozowski wrote:
>
>> Thank you Damian, it was very helpful.
>> I have implemented my solution in version 0.11.0.2 but there is one thing
>> I
>> still wonder.
>> So what I try to do is what is described in KIP 150. Since it didn't make
>> to the release for 1.0 I do it the old fashioned way.
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-150+-+
>> Kafka-Streams+Cogroup
>> First
>> KTable<K, V1> table1 =
>> builder.stream("topic1").groupByKey().aggregate(initializer1,
>> aggregator1, aggValueSerde1, storeName1);
>>
>>
>> for all the four topics and then I join the results.
>> And here is the thing, the topics are partitioned and I don't used global
>> tables, nor keyed messages and it seems to work fine.
>>
>> From Confluents documentation one could get impression that when reading
>> from partitoned topics you need to use global tables. But is it really
>> necessary in this case?
>> And if not then why?
>>
>> Thanks again
>> Artur
>>
>> Here is the link to my implementation
>>
>> https://github.com/afuyo/KStreamsDemo/blob/master/src/main/
>> java/kstream.demo/CustomerStreamPipelineHDI.java
>>
>> On Wed, Nov 22, 2017 at 12:10 PM, Damian Guy <da...@gmail.com>
>> wrote:
>>
>> Hi Artur,
>>>
>>> KafkaStreams 0.10.0.0 is quite old and a lot has changed and been fixed
>>> since then. If possible i'd recommend upgrading to at least 0.11.0.2 or
>>> 1.0.
>>> For joins you need to ensure that the topics have the same number of
>>> partitions (which they do) and that they are keyed the same.
>>>
>>> Thanks,
>>> Damian
>>>
>>> On Wed, 22 Nov 2017 at 11:02 Artur Mrozowski <ar...@gmail.com> wrote:
>>>
>>> Hi,
>>>> I am joining 4 different topic with 4 partitions each using 0.10.0.0
>>>> version of Kafka Streams. The joins are KTable to KTable. Is there
>>>> anything I should be aware of considering partitions or version of Kafka
>>>> Streams? In other words should I be expecting consistent results or do I
>>>> need to for example use Global tables.
>>>>
>>>> I'd like to run that application on Kubernetes later on. Should I think
>>>>
>>> of
>>>
>>>> anything or do different instances of the same Kafka Streams application
>>>> take care of management of the state?
>>>>
>>>> Grateful for any thoughts or a piece of advice
>>>>
>>>> Best Regards
>>>> /Artur
>>>>
>>>>
>
Re: Joins in Kafka Streams and partitioning of the topics
Posted by Jan Filipiak <Ja...@trivago.com>.
Hi,
Haven't checked your code. But from what you describe you should be fine.
Upgrading the version might help here and there but should still work
with 0.10
I guess.
Best Jan
On 30.11.2017 19:16, Artur Mrozowski wrote:
> Thank you Damian, it was very helpful.
> I have implemented my solution in version 0.11.0.2 but there is one thing I
> still wonder.
> So what I try to do is what is described in KIP 150. Since it didn't make
> to the release for 1.0 I do it the old fashioned way.
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-150+-+Kafka-Streams+Cogroup
> First
> KTable<K, V1> table1 =
> builder.stream("topic1").groupByKey().aggregate(initializer1,
> aggregator1, aggValueSerde1, storeName1);
>
>
> for all the four topics and then I join the results.
> And here is the thing, the topics are partitioned and I don't used global
> tables, nor keyed messages and it seems to work fine.
>
> From Confluents documentation one could get impression that when reading
> from partitoned topics you need to use global tables. But is it really
> necessary in this case?
> And if not then why?
>
> Thanks again
> Artur
>
> Here is the link to my implementation
>
> https://github.com/afuyo/KStreamsDemo/blob/master/src/main/java/kstream.demo/CustomerStreamPipelineHDI.java
>
> On Wed, Nov 22, 2017 at 12:10 PM, Damian Guy <da...@gmail.com> wrote:
>
>> Hi Artur,
>>
>> KafkaStreams 0.10.0.0 is quite old and a lot has changed and been fixed
>> since then. If possible i'd recommend upgrading to at least 0.11.0.2 or
>> 1.0.
>> For joins you need to ensure that the topics have the same number of
>> partitions (which they do) and that they are keyed the same.
>>
>> Thanks,
>> Damian
>>
>> On Wed, 22 Nov 2017 at 11:02 Artur Mrozowski <ar...@gmail.com> wrote:
>>
>>> Hi,
>>> I am joining 4 different topic with 4 partitions each using 0.10.0.0
>>> version of Kafka Streams. The joins are KTable to KTable. Is there
>>> anything I should be aware of considering partitions or version of Kafka
>>> Streams? In other words should I be expecting consistent results or do I
>>> need to for example use Global tables.
>>>
>>> I'd like to run that application on Kubernetes later on. Should I think
>> of
>>> anything or do different instances of the same Kafka Streams application
>>> take care of management of the state?
>>>
>>> Grateful for any thoughts or a piece of advice
>>>
>>> Best Regards
>>> /Artur
>>>
Re: Joins in Kafka Streams and partitioning of the topics
Posted by Artur Mrozowski <ar...@gmail.com>.
Thank you Damian, it was very helpful.
I have implemented my solution in version 0.11.0.2 but there is one thing I
still wonder.
So what I try to do is what is described in KIP 150. Since it didn't make
to the release for 1.0 I do it the old fashioned way.
https://cwiki.apache.org/confluence/display/KAFKA/KIP-150+-+Kafka-Streams+Cogroup
First
KTable<K, V1> table1 =
builder.stream("topic1").groupByKey().aggregate(initializer1,
aggregator1, aggValueSerde1, storeName1);
for all the four topics and then I join the results.
And here is the thing, the topics are partitioned and I don't used global
tables, nor keyed messages and it seems to work fine.
From Confluents documentation one could get impression that when reading
from partitoned topics you need to use global tables. But is it really
necessary in this case?
And if not then why?
Thanks again
Artur
Here is the link to my implementation
https://github.com/afuyo/KStreamsDemo/blob/master/src/main/java/kstream.demo/CustomerStreamPipelineHDI.java
On Wed, Nov 22, 2017 at 12:10 PM, Damian Guy <da...@gmail.com> wrote:
> Hi Artur,
>
> KafkaStreams 0.10.0.0 is quite old and a lot has changed and been fixed
> since then. If possible i'd recommend upgrading to at least 0.11.0.2 or
> 1.0.
> For joins you need to ensure that the topics have the same number of
> partitions (which they do) and that they are keyed the same.
>
> Thanks,
> Damian
>
> On Wed, 22 Nov 2017 at 11:02 Artur Mrozowski <ar...@gmail.com> wrote:
>
> > Hi,
> > I am joining 4 different topic with 4 partitions each using 0.10.0.0
> > version of Kafka Streams. The joins are KTable to KTable. Is there
> > anything I should be aware of considering partitions or version of Kafka
> > Streams? In other words should I be expecting consistent results or do I
> > need to for example use Global tables.
> >
> > I'd like to run that application on Kubernetes later on. Should I think
> of
> > anything or do different instances of the same Kafka Streams application
> > take care of management of the state?
> >
> > Grateful for any thoughts or a piece of advice
> >
> > Best Regards
> > /Artur
> >
>
Re: Joins in Kafka Streams and partitioning of the topics
Posted by Damian Guy <da...@gmail.com>.
Hi Artur,
KafkaStreams 0.10.0.0 is quite old and a lot has changed and been fixed
since then. If possible i'd recommend upgrading to at least 0.11.0.2 or 1.0.
For joins you need to ensure that the topics have the same number of
partitions (which they do) and that they are keyed the same.
Thanks,
Damian
On Wed, 22 Nov 2017 at 11:02 Artur Mrozowski <ar...@gmail.com> wrote:
> Hi,
> I am joining 4 different topic with 4 partitions each using 0.10.0.0
> version of Kafka Streams. The joins are KTable to KTable. Is there
> anything I should be aware of considering partitions or version of Kafka
> Streams? In other words should I be expecting consistent results or do I
> need to for example use Global tables.
>
> I'd like to run that application on Kubernetes later on. Should I think of
> anything or do different instances of the same Kafka Streams application
> take care of management of the state?
>
> Grateful for any thoughts or a piece of advice
>
> Best Regards
> /Artur
>