Hey Garry, I've opened:
https://issues.apache.org/jira/browse/SAMZA-144 To track the stream-level Kafka configuration override issue. Let me know if the "smallest" setting works for you. Cheers, Chris On 2/12/14 11:21 AM, "Chris Riccomini" <criccom...@linkedin.com> wrote: >Hey Garry, > >I just noticed that we don't actually support stream-level overrides for >Kafka configs. > >To tes this, you'll have to set the consumer settings at the system level: > >systems.kafka.consumer.auto.offset.reset=smallest > > >Note that this will cause you to read all data from kafka.tweets the first >time you run your job, as well. I think this is probably what you want, >but if not, you'd have to define two systems: one for the bootstrap >streams, and one for the tweet stream, so that you could configure the >bootstrap system to have the "smallest" reset setting. > >Cheers, >Chris > >On 2/12/14 11:06 AM, "Chris Riccomini" <criccom...@linkedin.com> wrote: > >>Hey Garry, >> >>I believe this is a similar issue to SAMZA-142. >> >>Can you try adding a config to set auto.offset.reset to smallest? >>Something like: >> >> >>systems.kafka.streams.positive-words.consumer.auto.offset.reset=smallest >> >>systems.kafka.streams.negative-words.consumer.auto.offset.reset=smallest >> >> >>This should change this log line: >> >>Final offset to be returned for Topic and Partition [positive-words,0] = >>2006 >> >> >>To something like: >> >>Final offset to be returned for Topic and Partition [positive-words,0] = >>0 >> >> >>Cheers, >>Chris >> >>On 2/12/14 10:44 AM, "Garry Turkington" <g.turking...@improvedigital.com> >>wrote: >> >>>Hi Chris, >>> >>>Sorry for the wrong log file! >>> >>>Samza container log is at: >>>http://pastebin.com/D5bAJd7U >>> >>>I do notice that it mentions returning the highest offset for the >>>supposedly bootstrapped streams which I presume shouldn't be happening. >>> >>>Thanks, >>>Garry >>> >>>-----Original Message----- >>>From: Chris Riccomini [mailto:criccom...@linkedin.com] >>>Sent: 12 February 2014 17:42 >>>To: dev@samza.incubator.apache.org >>>Subject: Re: Using bootstrap streams >>> >>>Hey Garry, >>> >>>So far, everything looks normal. >>> >>>The container you log you sent me actually appears to be the output of >>>the run-job.sh command. Since you're using YarnJobFactory, this is not >>>actually the container log. Could you grab the log from the container >>>that's running in YARN, and stick that in pastebin? You can usually find >>>this by going to YARN's RM (http://localhost:8088) and finding the link >>>to your ApplicationMaster. This will link to the logs for each container >>>that's running your tasks. >>> >>>Cheers, >>>Chris >>> >>>On 2/11/14 2:11 PM, "Garry Turkington" <g.turking...@improvedigital.com> >>>wrote: >>> >>>>Hi Chris, >>>> >>>>Following up on this, sorry for the delay, travelling this week. >>>> >>>>Main task config: >>>>http://pastebin.com/enQXLcbZ >>>> >>>>Container log: >>>>http://pastebin.com/YLiKp0CS >>>> >>>>I'm putting the positive and negative words into the bootstrap streams >>>>prior to running the job -- and confirmed the data is in the Kafka >>>>stream via kafka-console-consumer.sh with the --from-beginning option. >>>> >>>>Thanks for any input! >>>>Garry >>>> >>>>-----Original Message----- >>>>From: Chris Riccomini [mailto:criccom...@linkedin.com] >>>>Sent: 10 February 2014 23:25 >>>>To: dev@samza.incubator.apache.org >>>>Subject: Re: Using bootstrap streams >>>> >>>>Hey Garry, >>>> >>>>It sounds like your understanding of bootstrap streams is correct. >>>> >>>>Bootstrap stream messages will be delivered to the process() method >>>>just like any other. The only difference is you're supposed to get all >>>>of them from 0-lastOffset before you get any messages from >>>>non-bootstrap >>>>streams. >>>>Your positive/negative example sounds like a reasonable use case for a >>>>bootstrap stream. >>>> >>>>A few questions: >>>> >>>>1. Can you post the container logs and the full configuration file for >>>>your job somewhere (e.g. Github gist)? >>>>2. Are you putting data into the positive-words and negative-words >>>>topic before you start the Samza job? >>>> >>>>Also, you can do envelope.getSystemStreamPartition().getStream() >>>>directly (no need to call getSystemStream()). >>>> >>>>Cheers, >>>>Chris >>>> >>>>On 2/10/14 3:18 AM, "Garry Turkington" >>>><g.turking...@improvedigital.com> >>>>wrote: >>>> >>>>>Hi, >>>>> >>>>>I was building a task to do some sentiment analysis on incoming data. >>>>>I have a corpus each of positive and negative words to which the task >>>>>needs access. This seemed like a good fit for bootstrap streams. But I >>>>>can't seem to get them to work. >>>>> >>>>>I have my job configured with the 3 Kafka topics in task.inputs and >>>>>that seems to work, just throwing data at any of the topics is hitting >>>>>the task. >>>>> >>>>>But setting up the 2 reference streams as bootstrap doesn't seem to be >>>>>working. Here's the relevant part of the config, I want to read the >>>>>entire message history each time: >>>>> >>>>>systems.kafka.streams.positive-words.samza.bootstrap=true >>>>>systems.kafka.streams.positive-words.samza.reset.offset=true >>>>> >>>>>systems.kafka.streams.negative-words.samza.bootstrap=true >>>>>systems.kafka.streams.negative-words.samza.reset.offset=true >>>>> >>>>>Do bootstrap streams get handled in any special way, I'm assuming here >>>>>that the messages will arrive in the process method on StreamTask just >>>>>like any other and I can handle them differently by switching on >>>>>envelope.getSystemStreamPartition().getSystemStream().getStream(). >>>>>Looking at the code it looks the same with the BootstrapChooser doing >>>>>its magic to determine which message is delivered to the task but the >>>>>actual delivery seems the same. >>>>> >>>>>What am I missing? >>>>> >>>>>Thanks, >>>>>Garry >>>>> >>>> >>>> >>>>----- >>>>No virus found in this message. >>>>Checked by AVG - www.avg.com >>>>Version: 2014.0.4259 / Virus Database: 3697/7081 - Release Date: >>>>02/10/14 >>> >>> >>>----- >>>No virus found in this message. >>>Checked by AVG - www.avg.com >>>Version: 2014.0.4259 / Virus Database: 3697/7081 - Release Date: >>>02/10/14 >> >
