It works after calling through() before countByKey, so many 0.10.0.1
examples on the web missing the `through()` call and it will fail to get
the right output when running with input topic > 1 partitions.
Thanks very much all ! Finally got the correct results.
On Thu, Sep 1, 2016 at 4:52 AM, Matthias J. Sax <matth...@confluent.io>
wrote:
> Hi Tommy,
>
> I did checkout your github project and can verify the "issue". As you
> are using Kafka 0.10.0.1 the automatic repartitioning step is not
> available.
>
> If you use "trunk" version, your program will run as expected. If you
> want to stay with 0.10.0.1, you need to repartition the data after map()
> explicitly, via a call to through():
>
> > val wordCounts: KStream[String, JLong] = textLines
> > .flatMapValues(_.toLowerCase.split("\\W+").toIterable.asJava)
> > .map((key: String, word: String) => new KeyValue(word, word))
> > .through("my-repartitioing-topic")
> > .countByKey("counts")
> > .toStream
>
> Keep in mind, that it is recommended to create all user topics manually.
> Thus, you should create your repartitioning topic you specify in
> through() before you start your Kafka Streams application.
>
>
> -Matthias
>
>
> On 08/31/2016 09:07 PM, Guozhang Wang wrote:
> > Hello Tommy,
> >
> > Which version of Kafka are you using?
> >
> > Guozhang
> >
> > On Wed, Aug 31, 2016 at 4:41 AM, Tommy Q <deeplam...@gmail.com> wrote:
> >
> >> I cleaned up all the zookeeper & kafka states and run the WordCountDemo
> >> again, the results in wc-out is still wrong:
> >>
> >> a 1
> >>> b 1
> >>> a 1
> >>> b 1
> >>> c 1
> >>
> >>
> >>
> >> On Wed, Aug 31, 2016 at 5:32 PM, Michael Noll <mich...@confluent.io>
> >> wrote:
> >>
> >>> Can you double-check whether the results in wc-out are not rather:
> >>>
> >>> a 1
> >>> b 1
> >>> a 2
> >>> b 2
> >>> c 1
> >>>
> >>> ?
> >>>
> >>> On Wed, Aug 31, 2016 at 5:47 AM, Tommy Q <deeplam...@gmail.com> wrote:
> >>>
> >>>> Tried the word count example as discussed, the result in wc-out is
> >> wrong:
> >>>>
> >>>> a 1
> >>>>> b 1
> >>>>> a 1
> >>>>> b 1
> >>>>> c 1
> >>>>
> >>>>
> >>>> The expected result should be:
> >>>>
> >>>> a 2
> >>>>> b 2
> >>>>> c 1
> >>>>
> >>>>
> >>>> Kafka version is 0.10.0.1
> >>>>
> >>>>
> >>>> On Tue, Aug 30, 2016 at 10:29 PM, Matthias J. Sax <
> >> matth...@confluent.io
> >>>>
> >>>> wrote:
> >>>>
> >>>>> No. It does not support hidden topics.
> >>>>>
> >>>>> The only explanation might be, that there is no repartitioning step.
> >>> But
> >>>>> than the question would be, if there is a bug in Kafka Streams,
> >> because
> >>>>> between map() and countByKey() repartitioning is required.
> >>>>>
> >>>>> Can you verify that the result is correct?
> >>>>>
> >>>>> -Matthias
> >>>>>
> >>>>> On 08/30/2016 03:24 PM, Tommy Q wrote:
> >>>>>> Does Kafka support hidden topics ? (Since all the topics infos are
> >>>> stored
> >>>>>> in ZK, this probably not the case )
> >>>>>>
> >>>>>> On Tue, Aug 30, 2016 at 5:58 PM, Matthias J. Sax <
> >>>> matth...@confluent.io>
> >>>>>> wrote:
> >>>>>>
> >>>>>>> Hi Tommy,
> >>>>>>>
> >>>>>>> yes, you do understand Kafka Streams correctly. And yes, for
> >>>> shuffling,
> >>>>>>> na internal topic will be created under the hood. It should be
> >> named
> >>>>>>> "<application-id>-something-repartition". I am not sure, why it
> >> is
> >>>> not
> >>>>>>> listed via bin/kafka-topics.sh
> >>>>>>>
> >>>>>>> The internal topic "<application-id>-counts-changelog" you see is
> >>>>>>> created to back the state of countByKey() operator.
> >>>>>>>
> >>>>>>> See
> >>>>>>> https://cwiki.apache.org/confluence/display/KAFKA/
> >>>>>>> Kafka+Streams%3A+Internal+Data+Management
> >>>>>>>
> >>>>>>> and
> >>>>>>>
> >>>>>>> http://www.confluent.io/blog/data-reprocessing-with-kafka-
> >>>>>>> streams-resetting-a-streams-application
> >>>>>>>
> >>>>>>>
> >>>>>>> -Matthias
> >>>>>>>
> >>>>>>>
> >>>>>>> On 08/30/2016 06:55 AM, Tommy Q wrote:
> >>>>>>>> Michael, Thanks for your help.
> >>>>>>>>
> >>>>>>>> Take the word count example, I am trying to walk through the code
> >>>> based
> >>>>>>> on
> >>>>>>>> your explanation:
> >>>>>>>>
> >>>>>>>> val textLines: KStream[String, String] =
> >>>>>>> builder.stream("input-topic")
> >>>>>>>> val wordCounts: KStream[String, JLong] = textLines
> >>>>>>>> .flatMapValues(_.toLowerCase.split("\\W+").toIterable.
> >>> asJava)
> >>>>>>>> .map((key: String, word: String) => new KeyValue(word,
> >> word))
> >>>>>>>> .countByKey("counts")
> >>>>>>>> .toStream
> >>>>>>>>
> >>>>>>>> wordCounts.to(stringSerde, longSerde, "wc-out")
> >>>>>>>>
> >>>>>>>> Suppose the input-topic has two partitions and each partition
> >> has a
> >>>>>>> string
> >>>>>>>> record produced into:
> >>>>>>>>
> >>>>>>>> input-topic_0 : "a b"
> >>>>>>>>> input-topic_1 : "a b c"
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> Suppose we started two instance of the stream topology ( task_0
> >> and
> >>>>>>>> task_1). So after flatMapValues & map executed, they should have
> >>> the
> >>>>>>>> following task state:
> >>>>>>>>
> >>>>>>>> task_0 : [ (a, "a"), (b, "b") ]
> >>>>>>>>> task_1 : [ (a, "a"), (b: "b"), (c: "c") ]
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> Before the execution of countByKey, the kafka-stream framework
> >>>> should
> >>>>>>>> insert a invisible shuffle phase internally:
> >>>>>>>>
> >>>>>>>> shuffled across the network :
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>> _internal_topic_shuffle_0 : [ (a, "a"), (a, "a") ]
> >>>>>>>>> _internal_topic_shuffle_1 : [ (b, "b"), (b: "b"), (c: "c") ]
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> countByKey (reduce) :
> >>>>>>>>
> >>>>>>>> task_0 (counts-changelog_0) : [ (a, 2) ]
> >>>>>>>>
> >>>>>>>> task_1 (counts-changelog_1): [ (b, 2), (c, 1) ]
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> And after the execution of `wordCounts.to(stringSerde, longSerde,
> >>>>>>>> "wc-out")`, we get the word count output in wc-out topic:
> >>>>>>>>
> >>>>>>>> task_0 (wc-out_0) : [ (a, 2) ]
> >>>>>>>>
> >>>>>>>> task_1 (wc-out_1): [ (b, 2), (c, 1) ]
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> According the steps list above, do I understand the internals of
> >>>>> kstream
> >>>>>>>> word count correctly ?
> >>>>>>>> Another question is does the shuffle across the network work by
> >>>>> creating
> >>>>>>>> intermediate topics ? If so, why can't I find the intermediate
> >>> topics
> >>>>>>> using
> >>>>>>>> `bin/kafka-topics.sh --list --zookeeper localhost:2181` ? I can
> >>> only
> >>>>> see
> >>>>>>>> the counts-changelog got created by the kstream framework.
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> On Tue, Aug 30, 2016 at 2:25 AM, Michael Noll <
> >>> mich...@confluent.io>
> >>>>>>> wrote:
> >>>>>>>>
> >>>>>>>>> In Kafka Streams, data is partitioned according to the keys of
> >> the
> >>>>>>>>> key-value records, and operations such as countByKey operate on
> >>>> these
> >>>>>>>>> stream partitions. When reading data from Kafka, these stream
> >>>>>>> partitions
> >>>>>>>>> map to the partitions of the Kafka input topic(s), but these may
> >>>>> change
> >>>>>>>>> once you add processing operations.
> >>>>>>>>>
> >>>>>>>>> To your question: The first step, if the data isn't already
> >> keyed
> >>>> as
> >>>>>>>>> needed, is to select the key you want to count by, which results
> >>> in
> >>>> 1+
> >>>>>>>>> output stream partitions. Here, data may get shuffled across
> >> the
> >>>>>>> network
> >>>>>>>>> (but if won't if there's no need to, e.g. when the data is
> >> already
> >>>>>>> keyed as
> >>>>>>>>> needed). Then the count operation is performed for each stream
> >>>>>>> partition,
> >>>>>>>>> which is similar to the sort-and-reduce phase in Hadoop.
> >>>>>>>>>
> >>>>>>>>> On Mon, Aug 29, 2016 at 5:31 PM, Tommy <deeplam...@gmail.com>
> >>>> wrote:
> >>>>>>>>>
> >>>>>>>>>> Hi,
> >>>>>>>>>>
> >>>>>>>>>> For "word count" example in Hadoop, there are
> >>>> shuffle-sort-and-reduce
> >>>>>>>>>> phases that handles outputs from different mappers, how does it
> >>>> work
> >>>>> in
> >>>>>>>>>> KStream ?
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>>
> >>>>
> >>>
> >>
> >
> >
> >
>
>
