spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Dibyendu Bhattacharya <dibyendu.bhattach...@gmail.com>
Subject Re: Low Level Kafka Consumer for Spark
Date Wed, 10 Sep 2014 16:12:14 GMT
Hi ,

The latest changes with Kafka message re-play by manipulating ZK offset
seems to be working fine for us. This gives us some relief till actual
issue is fixed in Spark 1.2 .

I have some question on how Spark process the Received data. The logic I
used is basically to pull messages form individual partitions using
dedicated Receivers, and doing a Union of these Stream . After that I
process this union stream.

Today I wanted to test this consumer with our Internal Kafka cluster which
has around 50 million records, with this huge backlog I found Spark only
running the Receiver task and not running the Processing task (or rather
doing it very slow) . Is this a issue with the Consumer or it is a issue
from Spark side ? Ideally when Receivers durably write data to "Store" ,
the processing should start in parallel . Why does the processing task need
to wait till the Receiver consumes all 50 million messages. ...Or may be I
am doing something wrong ? I can share the driver log if you want.

in Driver I can see only "storage.BlockManagerInfo: Added input..." type
messages, but hardly I see "scheduler.TaskSetManager: Starting task..."
messages.. I see data getting written to target system in very very slow
pace.


Regards,
Dibyendu






On Mon, Sep 8, 2014 at 12:08 AM, Dibyendu Bhattacharya <
dibyendu.bhattachary@gmail.com> wrote:

> Hi Tathagata,
>
> I have managed to implement the logic into the Kafka-Spark consumer to
> recover from Driver failure. This is just a interim fix till actual fix is
> done from Spark side.
>
> The logic is something like this.
>
> 1. When the Individual Receivers starts for every Topic partition, it
> writes the Kafka messages along with certain meta data in Block Store. This
> meta data contains the details of message offset, partition id, topic name
> and consumer id. You can see this logic in PartitionManager.java  next()
> method.
>
> 2.  In the Driver code ( Consumer.java) , I am creating the union of all
> there individual D-Streams, and processing the data using forEachRDD call.
> In the driver code, I am receiving the RDD which contains the Kafka
> messages along with meta data details. In the driver code, periodically I
> am committing the "processed" offset of the Kafka message into ZK.
>
> 3. When driver stops, and restart again, the Receiver starts again, and
> this time in PartitionManager.java, I am checking what is the actual
> "committed" offset for the partition, and what is the actual "processed"
> offset of the same partition. This logic is in the PartitionManager
> constructor.
>
> If this is a Receiver restart, and "processed" offset of less than
> "Committed" offset, I am started fetching again from "Processed" offset.
> This may lead to duplicate records, but our system can handle duplicates.
>
> I have tested with multiple driver kill/stops and I found no data loss in
> Kafka consumer.
>
> In the Driver code, I have not done any "checkpointing" yet, will test
> that tomorrow.
>
>
> One interesting thing I found, if I do "repartition" of original stream ,
> I can still see the issue of data loss in this logic. What I believe,
> during re- partitioning Spark might be changing the order of RDDs the way
> it generated from Kafka stream. So during re-partition case, even when I am
> committing processed offset, but as this is not in order I still see issue.
> Not sure if this understanding is correct, but not able to find any other
> explanation.
>
> But if I do not use repartition this solution works fine.
>
> I can make this as configurable, so that when actual fix is available ,
> this feature in consumer can be turned off as this is an overhead for the
> consumer . Let me know what you think..
>
> Regards,
> Dibyendu
>
>
>
>
> On Fri, Sep 5, 2014 at 11:14 PM, Tathagata Das <
> tathagata.das1565@gmail.com> wrote:
>
>> Some thoughts on this thread to clarify the doubts.
>>
>> 1. Driver recovery: The current (1.1 to be released) does not recover the
>> raw data that has been received but not processes. This is because when the
>> driver dies, the executors die and so does the raw data that was stored in
>> it. Only for HDFS, the data is not lost by driver recovery as the data is
>> already present reliably in HDFS. This is something we want to fix by Spark
>> 1.2 (3 month from now). Regarding recovery by replaying the data from
>> Kafka, it is possible but tricky. Our goal is to provide strong guarantee,
>> exactly-once semantics in all transformations. To guarantee this for all
>> kinds of streaming computations stateful and not-stateful computations, it
>> is requires that the data be replayed through Kafka in exactly same order,
>> and the underlying blocks of data in Spark be regenerated in the exact way
>> as it would have if there was no driver failure. This is quite tricky to
>> implement, requires manipulation of zookeeper offsets, etc, that is hard to
>> do with the high level consumer that KafkaUtil uses. Dibyendu's low level
>> Kafka receiver may enable such approaches in the future. For now we
>> definitely plan to solve the first problem very very soon.
>>
>> 3. Repartitioning: I am trying to understand the repartition issue. One
>> common mistake I have seen is that developers repartition a stream but not
>> use the repartitioned stream.
>>
>> WRONG:
>> inputDstream.repartition(100)
>> inputDstream.map(...).count().print()
>>
>> RIGHT:
>> val repartitionedDStream = inputDStream.repartitoin(100)
>> repartitionedDStream.map(...).count().print()
>>
>> Not sure if this helps solve the problem that you all the facing. I am
>> going to add this to the stremaing programming guide to make sure this
>> common mistake is avoided.
>>
>> TD
>>
>>
>>
>>
>> On Wed, Sep 3, 2014 at 10:38 AM, Dibyendu Bhattacharya <
>> dibyendu.bhattachary@gmail.com> wrote:
>>
>>> Hi,
>>>
>>> Sorry for little delay . As discussed in this thread, I have modified
>>> the Kafka-Spark-Consumer (
>>> https://github.com/dibbhatt/kafka-spark-consumer) code to have
>>> dedicated Receiver for every Topic Partition. You can see the example howto
>>> create Union of these receivers in consumer.kafka.client.Consumer.java .
>>>
>>> Thanks to Chris for suggesting this change.
>>>
>>> Regards,
>>> Dibyendu
>>>
>>>
>>> On Mon, Sep 1, 2014 at 2:55 AM, RodrigoB <rodrigo.boavida@aspect.com>
>>> wrote:
>>>
>>>> Just a comment on the recovery part.
>>>>
>>>> Is it correct to say that currently Spark Streaming recovery design
>>>> does not
>>>> consider re-computations (upon metadata lineage recovery) that depend on
>>>> blocks of data of the received stream?
>>>> https://issues.apache.org/jira/browse/SPARK-1647
>>>>
>>>> Just to illustrate a real use case (mine):
>>>> - We have object states which have a Duration field per state which is
>>>> incremented on every batch interval. Also this object state is reset to
>>>> 0
>>>> upon incoming state changing events. Let's supposed there is at least
>>>> one
>>>> event since the last data checkpoint. This will lead to inconsistency
>>>> upon
>>>> driver recovery: The Duration field will get incremented from the data
>>>> checkpoint version until the recovery moment, but the state change event
>>>> will never be re-processed...so in the end we have the old state with
>>>> the
>>>> wrong Duration value.
>>>> To make things worst, let's imagine we're dumping the Duration increases
>>>> somewhere...which means we're spreading the problem across our system.
>>>> Re-computation awareness is something I've commented on another thread
>>>> and
>>>> rather treat it separately.
>>>>
>>>> http://apache-spark-user-list.1001560.n3.nabble.com/Spark-Streaming-checkpoint-recovery-causes-IO-re-execution-td12568.html#a13205
>>>>
>>>> Re-computations do occur, but the only RDD's that are recovered are the
>>>> ones
>>>> from the data checkpoint. This is what we've seen. Is not enough by
>>>> itself
>>>> to ensure recovery of computed data and this partial recovery leads to
>>>> inconsistency in some cases.
>>>>
>>>> Roger - I share the same question with you - I'm just not sure if the
>>>> replicated data really gets persisted on every batch. The execution
>>>> lineage
>>>> is checkpointed, but if we have big chunks of data being consumed to
>>>> Receiver node on let's say a second bases then having it persisted to
>>>> HDFS
>>>> every second could be a big challenge for keeping JVM performance -
>>>> maybe
>>>> that could be reason why it's not really implemented...assuming it
>>>> isn't.
>>>>
>>>> Dibyendu had a great effort with the offset controlling code but the
>>>> general
>>>> state consistent recovery feels to me like another big issue to address.
>>>>
>>>> I plan on having a dive into the Streaming code and try to at least
>>>> contribute with some ideas. Some more insight from anyone on the dev
>>>> team
>>>> will be very appreciated.
>>>>
>>>> tnks,
>>>> Rod
>>>>
>>>>
>>>>
>>>>
>>>> --
>>>> View this message in context:
>>>> http://apache-spark-user-list.1001560.n3.nabble.com/Low-Level-Kafka-Consumer-for-Spark-tp11258p13208.html
>>>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>>>
>>>> ---------------------------------------------------------------------
>>>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>>>> For additional commands, e-mail: user-help@spark.apache.org
>>>>
>>>>
>>>
>>
>

Mime
View raw message