kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Matthias J. Sax" <matth...@confluent.io>
Subject Re: Debugging Kafka Streams Windowing
Date Wed, 03 May 2017 23:05:41 GMT
I would recommend to double check the following:

 - can you confirm that the filter does not remove all data for those
time periods?
 - I would also check input for your AggregatorFunction() -- does it
receive everything?
 - same for .mapValues()

This would help to understand in what part of the program the data gets
lost.


-Matthias


On 5/2/17 11:09 PM, Mahendra Kariya wrote:
> Hi Garrett,
> 
> Thanks for these insights. But we are not consuming old data. We want the
> Streams app to run in near real time. And that is how it is actually
> running. The lag never increases beyond a certain limit. So I don't think
> that's an issue.
> 
> The values of the configs that you are mentioning are whatever Kafka offers
> by default. So I guess that should be fine.
> 
> 
> 
> On Tue, May 2, 2017 at 7:52 PM, Garrett Barton <garrett.barton@gmail.com>
> wrote:
> 
>> Mahendra,
>>
>>  One possible thing I have seen that exhibits the same behavior of missing
>> windows of data is the configuration of the topics (internal and your own)
>> retention policies.  I was loading data that was fairly old (weeks) and
>> using event time semantics as the record timestamp (custom timestamp
>> extractor) and the cleanup stuff was deleting segments nearly right after
>> they were written.  In my case default cleanup run was every 5 minutes, and
>> the default retention was 7 days, so every 5 minutes I lost data.  In my
>> logs I saw a ton of warnings about 'offset not found' and kafka skipping
>> ahead to whatever the next available offset was.  End result was gaps all
>> over my data.  I don't have a good fix yet, I set the retention to
>> something massive which I think is getting me other problems.
>>
>> Maybe that helps?
>>
>> On Tue, May 2, 2017 at 6:27 AM, Mahendra Kariya <
>> mahendra.kariya@go-jek.com>
>> wrote:
>>
>>> Hi Matthias,
>>>
>>> What we did was read the data from sink topic and print it to console.
>> And
>>> here's the raw data from that topic (the counts are randomized). As we
>> can
>>> see, the data is certainly missing for some time windows. For instance,
>>> after 1493693760, the next timestamp for which the data is present
>>> is 1493694300. That's around 9 minutes of data missing.
>>>
>>> And this is just one instance. There are a lot of such instances in this
>>> file.
>>>
>>>
>>>
>>> On Sun, Apr 30, 2017 at 11:23 AM, Mahendra Kariya <
>>> mahendra.kariya@go-jek.com> wrote:
>>>
>>>> Thanks for the update Matthias! And sorry for the delayed response.
>>>>
>>>> The reason we use .aggregate() is because we want to count the number of
>>>> unique values for a particular field in the message. So, we just add
>> that
>>>> particular field's value in the HashSet and then take the size of the
>>>> HashSet.
>>>>
>>>> On our side, we are also investigating and it looks like there might be
>> a
>>>> bug somewhere in our codebase. If that's the case, then it's quite
>> possible
>>>> that there is no bug in Kafka Streams, except the metric one.
>>>>
>>>> We will revert after confirming.
>>>>
>>>>
>>>>
>>>>
>>>> On Sun, Apr 30, 2017 at 10:39 AM, Matthias J. Sax <
>> matthias@confluent.io>
>>>> wrote:
>>>>
>>>>> Just a follow up (we identified a bug in the "skipped records" metric).
>>>>> The reported value is not correct.
>>>>>
>>>>>
>>>>> On 4/28/17 9:12 PM, Matthias J. Sax wrote:
>>>>>> Ok. That makes sense.
>>>>>>
>>>>>> Question: why do you use .aggregate() instead of .count() ?
>>>>>>
>>>>>> Also, can you share the code of you AggregatorFunction()? Did you
>>>>> change
>>>>>> any default setting of StreamsConfig?
>>>>>>
>>>>>> I have still no idea what could go wrong. Maybe you can run with
log
>>>>>> level TRACE? Maybe we can get some insight from those.
>>>>>>
>>>>>>
>>>>>> -Matthias
>>>>>>
>>>>>> On 4/27/17 11:41 PM, Mahendra Kariya wrote:
>>>>>>> Oh good point!
>>>>>>>
>>>>>>> The reason why there is only one row corresponding to each time
>>>>> window is
>>>>>>> because it only contains the latest value for the time window.
So
>>>>> what we
>>>>>>> did was we just dumped the data present in the sink topic to
a db
>>>>> using an
>>>>>>> upsert query. The primary key of the table was time window. The
file
>>>>> that I
>>>>>>> attached is actually the data present in the DB. And we know
that
>>>>> there is
>>>>>>> no bug in our db dump code because we have been using it for
a long
>>>>> time in
>>>>>>> production without any issues.
>>>>>>>
>>>>>>> The reason the count is zero for some time windows is because
I
>>>>> subtracted
>>>>>>> a random number the actual values and rounded it off to zero;
for
>>>>> privacy
>>>>>>> reason. The actual data doesn't have any zero values. I should
have
>>>>>>> mentioned this earlier. My bad!
>>>>>>>
>>>>>>> The stream topology code looks something like this.
>>>>>>>
>>>>>>> stream
>>>>>>>     .filter()
>>>>>>>     .map((key, value) -> new KeyValue<>(transform(key),
value)
>>>>>>>     .groupByKey()
>>>>>>>     .aggregate(HashSet::new, AggregatorFunction(),
>>>>>>> TimeWindows.of(60000).until(3600000))
>>>>>>>     .mapValues(HashSet::size)
>>>>>>>     .toStream()
>>>>>>>     .map((key, value) -> convertToProtobufObject(key, value))
>>>>>>>     .to()
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> On Fri, Apr 28, 2017 at 1:13 PM, Matthias J. Sax <
>>>>> matthias@confluent.io>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Thanks for the details (sorry that I forgot that you did
share the
>>>>>>>> output already).
>>>>>>>>
>>>>>>>> Might be a dumb question, but what is the count for missing
windows
>>>>> in
>>>>>>>> your seconds implementation?
>>>>>>>>
>>>>>>>> If there is no data for a window, it should not emit a window
with
>>>>> count
>>>>>>>> zero, but nothing.
>>>>>>>>
>>>>>>>> Thus, looking at your output, I am wondering how it could
contain
>>>>> line
>>>>>>>> like:
>>>>>>>>
>>>>>>>>> 2017-04-27T04:53:00 0
>>>>>>>>
>>>>>>>> I am also wondering why your output only contains a single
value
>> per
>>>>>>>> window. As Streams outputs multiple updates per window while
the
>>>>> count
>>>>>>>> is increasing, you should actually see multiple records per
window.
>>>>>>>>
>>>>>>>> Your code is like this:
>>>>>>>>
>>>>>>>> stream.filter().groupByKey().count(TimeWindow.of(60000)).to();
>>>>>>>>
>>>>>>>> Or do you have something more complex?
>>>>>>>>
>>>>>>>>
>>>>>>>> -Matthias
>>>>>>>>
>>>>>>>>
>>>>>>>> On 4/27/17 9:16 PM, Mahendra Kariya wrote:
>>>>>>>>>> Can you somehow verify your output?
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> Do you mean the Kafka streams output? In the Kafka Streams
output,
>>>>> we do
>>>>>>>>> see some missing values. I have attached the Kafka Streams
output
>>>>> (for a
>>>>>>>>> few hours) in the very first email of this thread for
reference.
>>>>>>>>>
>>>>>>>>> Let me also summarise what we have done so far.
>>>>>>>>>
>>>>>>>>> We took a dump of the raw data present in the source
topic. We
>>>>> wrote a
>>>>>>>>> script to read this data and do the exact same aggregations
that
>> we
>>>>> do
>>>>>>>>> using Kafka Streams. And then we compared the output
from Kafka
>>>>> Streams
>>>>>>>> and
>>>>>>>>> our script.
>>>>>>>>>
>>>>>>>>> The difference that we observed in the two outputs is
that there
>>>>> were a
>>>>>>>> few
>>>>>>>>> rows (corresponding to some time windows) missing in
the Streams
>>>>> output.
>>>>>>>>> For the time windows for which the data was present,
the
>> aggregated
>>>>>>>> numbers
>>>>>>>>> matched exactly.
>>>>>>>>>
>>>>>>>>> This means, either all the records for a particular time
window
>> are
>>>>> being
>>>>>>>>> skipped, or none. Now this is highly unlikely to happen.
Maybe
>>>>> there is a
>>>>>>>>> bug somewhere in the rocksdb state stores? Just a speculation,
not
>>>>> sure
>>>>>>>>> though. And there could even be a bug in the reported
metric.
>>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>>
>>>>
>>>
>>
> 


Mime
View raw message