That's a great tip! Thank you On Wed, May 17, 2017 at 9:38 AM Matthias J. Sax <matth...@confluent.io> wrote:
> Hey, > > if you need to request topic creation in advance, I would recommend to > do manual re-partitioning via through() -- this allows you to control > the topic names and should make your setup more robust. > > Eg. > > stream.selectKey().through("my-own-topic-name").groupByKey()... > > For this case, Streams will not add an internal re-partitioning topic, > as the "re-partition required flag" gets cleared via `through()` and > thus `groupByKey()` knows that data is already partitioned correctly. > > > > -Matthias > > On 5/16/17 5:40 PM, João Peixoto wrote: > > Your explanation makes sense. If I understood correctly it means that one > > stream thread can actually generate records that will be aggregated in a > > different thread, based on the new partitioning. > > > > I didn't think of that case, which now makes more sense. > > In my particular case the keys just get appended with some extra > > information, so I know there is no need for repartitioning. E.g. "mykey" > > > > selectKey > "mykey:currentHour" (just an example, I'm not doing windowed > > operations). > > > > The processor API is always a possibility, this is not causing any > > performance issues whatsoever. Our Kafka Brokers do not allow automatic > > creation of topics so I actually need to request the creation of these > > internal topics. I know it is not recommended as the naming convention is > > not guaranteed to remain the same in future releases, but the security > > configuration there is not final as of now. > > > > Thanks for the info > > > > On Tue, May 16, 2017 at 5:28 PM Matthias J. Sax <matth...@confluent.io> > > wrote: > > > >> João, > >> > >> in your example, k.toUpperCase() does break partitioning. Assume you > >> have two records <a,5> and <A,10> -- both do have different keys and > >> might be contained in different partitions. After you do a selectKey(), > >> both do have the same key. In order to compute the aggregation > >> correctly, it is required to re-partition the data to make sure that > >> both records <A,5> and <A,10> are processed together. Otherwise, the > >> aggregation result would be incorrect. > >> > >> If and only if you know, that all key are in lowercase (or all keys are > >> in uppercase), the re-partitioning would not be required. But Streams > >> cannot know this and thus conservatively does re-partition to ensure > >> correctness. > >> > >> Note, that the `toUpperCase()` would not make sense if all keys are in > >> upper case already. Furthermore, if all keys are in lower case, you can > >> compute the aggregation on the lower case keys directly and convert the > >> keys of the result into upper case -- this would allow you to avoid the > >> re-partitioning topic. > >> > >> Does this make sense? > >> > >> In general, you should use `selectKey()`, `map()` etc only if you need > >> to set a new key and thus break partitioning. For you don't modify the > >> key, you should use `mapValues()` for example. > >> > >> > >> Nevertheless, there are still some cases, for which the actual key must > >> be modified before a key-based operation and user wants to suppress > >> re-partitioning as she knows that partitioning is preserved (cf. > >> https://issues.apache.org/jira/browse/KAFKA-4835). This is currently > not > >> supported at DSL level. However, you could fall back to Processor API if > >> this is really critical. In general, it seems to be a corner case > >> optimization though. > >> > >> > >> > >> -Matthias > >> > >> > >> On 5/16/17 4:44 PM, João Peixoto wrote: > >>> Certain operations require a repartition topic, such as "selectKey" or > >>> "map". What purpose serves this repartition topic? > >>> > >>> Sample record: {"key": "a", ...} > >>> > >>> Stream: source.selectKey((k, v) -> KeyValue.pair(k.toUpperCase(), > >>> v)).groupByKey() //... > >>> > >>> From my understanding, the repartition topic will guarantee that if we > >> are > >>> reading from partition N, the new key will be written to the same > >> partition > >>> N on the repartition topic, which allows the stream task to always > handle > >>> the same partition number all the way. > >>> > >>> This seems relevant if the topology above is followed by: > >>> /*...*/.toStream().leftJoin(kTable) //... > >>> We are still processing the same partition number. If the source stream > >> and > >>> the kTable are co-partitioned, so will be the repartition topic. > >>> > >>> However in cases where there are no other operations in the topology > like > >>> "joins", that repartition topic seems useless. > >>> > >>> There's a thread on this subject > >>> < > >> > http://mail-archives.apache.org/mod_mbox/kafka-users/201705.mbox/%3CCAJikTEUHR=r0ika6vlf_y+qajxg8f_q19og_-s+q-gozpqb...@mail.gmail.com%3E > >>> , > >>> specific to topics with one partition only. The argument there is that > >>> repartition does not make sense on a topic with 1 partition only. > >> However, > >>> even if you have multiple partitions but never join with anything else, > >> it > >>> may not make sense for the reasons above. > >>> > >> > >> > > > >