You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Neelesh <ne...@gmail.com> on 2015/09/25 19:50:01 UTC
Kafka & Spark Streaming
Hi,
We are using DirectKafkaInputDStream and store completed consumer
offsets in Kafka (0.8.2). However, some of our use case require that
offsets be not written if processing of a partition fails with certain
exceptions. This allows us to build various backoff strategies for that
partition, instead of either blindly committing consumer offsets regardless
of errors (because KafkaRDD as HasOffsetRanges is available only on the
driver) or relying on Spark's retry logic and continuing without remedial
action.
I was playing with SparkListener and found that while one can listen on
taskCompletedEvent on the driver and even figure out that there was an
error, there is no way of mapping this task back to the partition and
retrieving offset range, topic & kafka partition # etc.
Any pointers appreciated!
Thanks!
-neelesh
Re: Kafka & Spark Streaming
Posted by Petr Novak <os...@gmail.com>.
You can have offsetRanges on workers f.e.
object Something {
var offsetRanges = Array[OffsetRange]()
def create[F : ClassTag](stream: InputDStream[Array[Byte]])
(implicit codec: Codec[F]: DStream[F] = {
stream transform { rdd =>
offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd flatMap { message =>
Try(codec.decode(message)) match {
case Success(fact) => Some(fact)
case Failure(e) => None
}
}
}
}
call create and use returned stream downstream.
or something like
// See https://issues.apache.org/jira/browse/SPARK-5569 why I map
OffsetRamges to a custom class
case class TopicMetadata(name: String, partition: Int, fromOffset:
Long, untilOffset: Long)
object KafkaContext {
private[this] var state = Array[TopicMetadata]()
def captureTopicMetadata(offsetRanges: Array[OffsetRange]): Unit = {
state = offsetRanges.map { o =>
TopicMetadata(o.topic, o.partition, o.fromOffset, o.untilOffset)
}
}
def topics: Array[TopicMetadata] = state
}
//then somewhere
def run(steam) = {
stream.transform { rdd =>
KafkaContext.captureTopicMetadata(rdd.asInstanceOf[HasOffsetRanges].offsetRanges)
rdd
....
....
....
}
.forecahRDD {
val s = KafkaContext.topics.map { x =>
s"${x.name}_${x.partition}_${x.fromOffset}-${x.untilOffset}"
}
...
}
}
So they can be available on Driver. Sorry for not precise code. I'm in a
hurry. There a probably mistakes but you can get the idea.
Petr
On Fri, Sep 25, 2015 at 7:50 PM, Neelesh <ne...@gmail.com> wrote:
> Hi,
> We are using DirectKafkaInputDStream and store completed consumer
> offsets in Kafka (0.8.2). However, some of our use case require that
> offsets be not written if processing of a partition fails with certain
> exceptions. This allows us to build various backoff strategies for that
> partition, instead of either blindly committing consumer offsets regardless
> of errors (because KafkaRDD as HasOffsetRanges is available only on the
> driver) or relying on Spark's retry logic and continuing without remedial
> action.
>
> I was playing with SparkListener and found that while one can listen on
> taskCompletedEvent on the driver and even figure out that there was an
> error, there is no way of mapping this task back to the partition and
> retrieving offset range, topic & kafka partition # etc.
>
> Any pointers appreciated!
>
> Thanks!
> -neelesh
>
Re: Kafka & Spark Streaming
Posted by Neelesh <ne...@gmail.com>.
Thanks. Ill keep an eye on this. Our implementation of the DStream
basically accepts a function to compute current offsets. The implementation
of the function fetches list of topics from zookeeper once in while. It
then adds consumer offsets for newly added topics with the currentOffsets
thats in memory & deletes removed topics. The "once in a while" is
pluggable as well, and we are planning to use ZK watches instead of a time
based refresh. Works for us because we use ZK extensively for a lot of
other book keeping.
On Fri, Sep 25, 2015 at 1:16 PM, Cody Koeninger <co...@koeninger.org> wrote:
> Yes, the partition IDs are the same.
>
> As far as the failure / subclassing goes, you may want to keep an eye on
> https://issues.apache.org/jira/browse/SPARK-10320 , not sure if the
> suggestions in there will end up going anywhere.
>
> On Fri, Sep 25, 2015 at 3:01 PM, Neelesh <ne...@gmail.com> wrote:
>
>> For the 1-1 mapping case, can I use TaskContext.get().partitionId as an
>> index in to the offset ranges?
>> For the failure case, yes, I'm subclassing of DirectKafkaInputDStream.
>> As for failures, different partitions in the same batch may be talking to
>> different RDBMS servers due to multitenancy - a spark streaming app is
>> consuming from several topics, each topic mapped to a customer for example.
>> It is quite possible that in a batch, several partitions belonging to the
>> same customer may fail, and others will go through. We don't want the whole
>> job to be killed because of one failing customer,and affect others in the
>> same job. Hope that makes sense.
>>
>> thnx
>>
>> On Fri, Sep 25, 2015 at 12:52 PM, Cody Koeninger <co...@koeninger.org>
>> wrote:
>>
>>> Your success case will work fine, it is a 1-1 mapping as you said.
>>>
>>> To handle failures in exactly the way you describe, you'd need to
>>> subclass or modify DirectKafkaInputDStream and change the way compute()
>>> works.
>>>
>>> Unless you really are going to have very fine-grained failures (why
>>> would only a given partition be failing while the rest are fine?) it's
>>> going to be easier to just fail the whole task and retry, or eventually
>>> kill the job.
>>>
>>> On Fri, Sep 25, 2015 at 1:55 PM, Neelesh <ne...@gmail.com> wrote:
>>>
>>>> Thanks Petr, Cody. This is a reasonable place to start for me. What I'm
>>>> trying to achieve
>>>>
>>>> stream.foreachRDD {rdd=>
>>>> rdd.foreachPartition { p=>
>>>>
>>>> Try(myFunc(...)) match {
>>>> case Sucess(s) => updatewatermark for this partition //of
>>>> course, expectation is that it will work only if there is a 1-1 mapping at
>>>> this point in time
>>>> case Failure() => Tell the driver not to generate a partition
>>>> for this kafka topic+partition for a while, by updating some shared state
>>>> (zk)
>>>>
>>>> }
>>>>
>>>> }
>>>> }
>>>>
>>>> I was looking for that mapping b/w kafka partition thats bound to a
>>>> task inside the task execution code, in cases where the intermediate
>>>> operations do not change partitions, shuffle etc.
>>>>
>>>> -neelesh
>>>>
>>>> On Fri, Sep 25, 2015 at 11:14 AM, Cody Koeninger <co...@koeninger.org>
>>>> wrote:
>>>>
>>>>>
>>>>> http://spark.apache.org/docs/latest/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers
>>>>>
>>>>> also has an example of how to close over the offset ranges so they are
>>>>> available on executors.
>>>>>
>>>>> On Fri, Sep 25, 2015 at 12:50 PM, Neelesh <ne...@gmail.com> wrote:
>>>>>
>>>>>> Hi,
>>>>>> We are using DirectKafkaInputDStream and store completed consumer
>>>>>> offsets in Kafka (0.8.2). However, some of our use case require that
>>>>>> offsets be not written if processing of a partition fails with certain
>>>>>> exceptions. This allows us to build various backoff strategies for that
>>>>>> partition, instead of either blindly committing consumer offsets regardless
>>>>>> of errors (because KafkaRDD as HasOffsetRanges is available only on the
>>>>>> driver) or relying on Spark's retry logic and continuing without remedial
>>>>>> action.
>>>>>>
>>>>>> I was playing with SparkListener and found that while one can listen
>>>>>> on taskCompletedEvent on the driver and even figure out that there was an
>>>>>> error, there is no way of mapping this task back to the partition and
>>>>>> retrieving offset range, topic & kafka partition # etc.
>>>>>>
>>>>>> Any pointers appreciated!
>>>>>>
>>>>>> Thanks!
>>>>>> -neelesh
>>>>>>
>>>>>
>>>>>
>>>>
>>>
>>
>
Re: Kafka & Spark Streaming
Posted by Cody Koeninger <co...@koeninger.org>.
Yes, the partition IDs are the same.
As far as the failure / subclassing goes, you may want to keep an eye on
https://issues.apache.org/jira/browse/SPARK-10320 , not sure if the
suggestions in there will end up going anywhere.
On Fri, Sep 25, 2015 at 3:01 PM, Neelesh <ne...@gmail.com> wrote:
> For the 1-1 mapping case, can I use TaskContext.get().partitionId as an
> index in to the offset ranges?
> For the failure case, yes, I'm subclassing of DirectKafkaInputDStream. As
> for failures, different partitions in the same batch may be talking to
> different RDBMS servers due to multitenancy - a spark streaming app is
> consuming from several topics, each topic mapped to a customer for example.
> It is quite possible that in a batch, several partitions belonging to the
> same customer may fail, and others will go through. We don't want the whole
> job to be killed because of one failing customer,and affect others in the
> same job. Hope that makes sense.
>
> thnx
>
> On Fri, Sep 25, 2015 at 12:52 PM, Cody Koeninger <co...@koeninger.org>
> wrote:
>
>> Your success case will work fine, it is a 1-1 mapping as you said.
>>
>> To handle failures in exactly the way you describe, you'd need to
>> subclass or modify DirectKafkaInputDStream and change the way compute()
>> works.
>>
>> Unless you really are going to have very fine-grained failures (why would
>> only a given partition be failing while the rest are fine?) it's going to
>> be easier to just fail the whole task and retry, or eventually kill the job.
>>
>> On Fri, Sep 25, 2015 at 1:55 PM, Neelesh <ne...@gmail.com> wrote:
>>
>>> Thanks Petr, Cody. This is a reasonable place to start for me. What I'm
>>> trying to achieve
>>>
>>> stream.foreachRDD {rdd=>
>>> rdd.foreachPartition { p=>
>>>
>>> Try(myFunc(...)) match {
>>> case Sucess(s) => updatewatermark for this partition //of
>>> course, expectation is that it will work only if there is a 1-1 mapping at
>>> this point in time
>>> case Failure() => Tell the driver not to generate a partition
>>> for this kafka topic+partition for a while, by updating some shared state
>>> (zk)
>>>
>>> }
>>>
>>> }
>>> }
>>>
>>> I was looking for that mapping b/w kafka partition thats bound to a task
>>> inside the task execution code, in cases where the intermediate operations
>>> do not change partitions, shuffle etc.
>>>
>>> -neelesh
>>>
>>> On Fri, Sep 25, 2015 at 11:14 AM, Cody Koeninger <co...@koeninger.org>
>>> wrote:
>>>
>>>>
>>>> http://spark.apache.org/docs/latest/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers
>>>>
>>>> also has an example of how to close over the offset ranges so they are
>>>> available on executors.
>>>>
>>>> On Fri, Sep 25, 2015 at 12:50 PM, Neelesh <ne...@gmail.com> wrote:
>>>>
>>>>> Hi,
>>>>> We are using DirectKafkaInputDStream and store completed consumer
>>>>> offsets in Kafka (0.8.2). However, some of our use case require that
>>>>> offsets be not written if processing of a partition fails with certain
>>>>> exceptions. This allows us to build various backoff strategies for that
>>>>> partition, instead of either blindly committing consumer offsets regardless
>>>>> of errors (because KafkaRDD as HasOffsetRanges is available only on the
>>>>> driver) or relying on Spark's retry logic and continuing without remedial
>>>>> action.
>>>>>
>>>>> I was playing with SparkListener and found that while one can listen
>>>>> on taskCompletedEvent on the driver and even figure out that there was an
>>>>> error, there is no way of mapping this task back to the partition and
>>>>> retrieving offset range, topic & kafka partition # etc.
>>>>>
>>>>> Any pointers appreciated!
>>>>>
>>>>> Thanks!
>>>>> -neelesh
>>>>>
>>>>
>>>>
>>>
>>
>
Re: Kafka & Spark Streaming
Posted by Neelesh <ne...@gmail.com>.
For the 1-1 mapping case, can I use TaskContext.get().partitionId as an
index in to the offset ranges?
For the failure case, yes, I'm subclassing of DirectKafkaInputDStream. As
for failures, different partitions in the same batch may be talking to
different RDBMS servers due to multitenancy - a spark streaming app is
consuming from several topics, each topic mapped to a customer for example.
It is quite possible that in a batch, several partitions belonging to the
same customer may fail, and others will go through. We don't want the whole
job to be killed because of one failing customer,and affect others in the
same job. Hope that makes sense.
thnx
On Fri, Sep 25, 2015 at 12:52 PM, Cody Koeninger <co...@koeninger.org> wrote:
> Your success case will work fine, it is a 1-1 mapping as you said.
>
> To handle failures in exactly the way you describe, you'd need to subclass
> or modify DirectKafkaInputDStream and change the way compute() works.
>
> Unless you really are going to have very fine-grained failures (why would
> only a given partition be failing while the rest are fine?) it's going to
> be easier to just fail the whole task and retry, or eventually kill the job.
>
> On Fri, Sep 25, 2015 at 1:55 PM, Neelesh <ne...@gmail.com> wrote:
>
>> Thanks Petr, Cody. This is a reasonable place to start for me. What I'm
>> trying to achieve
>>
>> stream.foreachRDD {rdd=>
>> rdd.foreachPartition { p=>
>>
>> Try(myFunc(...)) match {
>> case Sucess(s) => updatewatermark for this partition //of
>> course, expectation is that it will work only if there is a 1-1 mapping at
>> this point in time
>> case Failure() => Tell the driver not to generate a partition
>> for this kafka topic+partition for a while, by updating some shared state
>> (zk)
>>
>> }
>>
>> }
>> }
>>
>> I was looking for that mapping b/w kafka partition thats bound to a task
>> inside the task execution code, in cases where the intermediate operations
>> do not change partitions, shuffle etc.
>>
>> -neelesh
>>
>> On Fri, Sep 25, 2015 at 11:14 AM, Cody Koeninger <co...@koeninger.org>
>> wrote:
>>
>>>
>>> http://spark.apache.org/docs/latest/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers
>>>
>>> also has an example of how to close over the offset ranges so they are
>>> available on executors.
>>>
>>> On Fri, Sep 25, 2015 at 12:50 PM, Neelesh <ne...@gmail.com> wrote:
>>>
>>>> Hi,
>>>> We are using DirectKafkaInputDStream and store completed consumer
>>>> offsets in Kafka (0.8.2). However, some of our use case require that
>>>> offsets be not written if processing of a partition fails with certain
>>>> exceptions. This allows us to build various backoff strategies for that
>>>> partition, instead of either blindly committing consumer offsets regardless
>>>> of errors (because KafkaRDD as HasOffsetRanges is available only on the
>>>> driver) or relying on Spark's retry logic and continuing without remedial
>>>> action.
>>>>
>>>> I was playing with SparkListener and found that while one can listen on
>>>> taskCompletedEvent on the driver and even figure out that there was an
>>>> error, there is no way of mapping this task back to the partition and
>>>> retrieving offset range, topic & kafka partition # etc.
>>>>
>>>> Any pointers appreciated!
>>>>
>>>> Thanks!
>>>> -neelesh
>>>>
>>>
>>>
>>
>
Re: Kafka & Spark Streaming
Posted by Cody Koeninger <co...@koeninger.org>.
Your success case will work fine, it is a 1-1 mapping as you said.
To handle failures in exactly the way you describe, you'd need to subclass
or modify DirectKafkaInputDStream and change the way compute() works.
Unless you really are going to have very fine-grained failures (why would
only a given partition be failing while the rest are fine?) it's going to
be easier to just fail the whole task and retry, or eventually kill the job.
On Fri, Sep 25, 2015 at 1:55 PM, Neelesh <ne...@gmail.com> wrote:
> Thanks Petr, Cody. This is a reasonable place to start for me. What I'm
> trying to achieve
>
> stream.foreachRDD {rdd=>
> rdd.foreachPartition { p=>
>
> Try(myFunc(...)) match {
> case Sucess(s) => updatewatermark for this partition //of course,
> expectation is that it will work only if there is a 1-1 mapping at this
> point in time
> case Failure() => Tell the driver not to generate a partition
> for this kafka topic+partition for a while, by updating some shared state
> (zk)
>
> }
>
> }
> }
>
> I was looking for that mapping b/w kafka partition thats bound to a task
> inside the task execution code, in cases where the intermediate operations
> do not change partitions, shuffle etc.
>
> -neelesh
>
> On Fri, Sep 25, 2015 at 11:14 AM, Cody Koeninger <co...@koeninger.org>
> wrote:
>
>>
>> http://spark.apache.org/docs/latest/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers
>>
>> also has an example of how to close over the offset ranges so they are
>> available on executors.
>>
>> On Fri, Sep 25, 2015 at 12:50 PM, Neelesh <ne...@gmail.com> wrote:
>>
>>> Hi,
>>> We are using DirectKafkaInputDStream and store completed consumer
>>> offsets in Kafka (0.8.2). However, some of our use case require that
>>> offsets be not written if processing of a partition fails with certain
>>> exceptions. This allows us to build various backoff strategies for that
>>> partition, instead of either blindly committing consumer offsets regardless
>>> of errors (because KafkaRDD as HasOffsetRanges is available only on the
>>> driver) or relying on Spark's retry logic and continuing without remedial
>>> action.
>>>
>>> I was playing with SparkListener and found that while one can listen on
>>> taskCompletedEvent on the driver and even figure out that there was an
>>> error, there is no way of mapping this task back to the partition and
>>> retrieving offset range, topic & kafka partition # etc.
>>>
>>> Any pointers appreciated!
>>>
>>> Thanks!
>>> -neelesh
>>>
>>
>>
>
Re: Kafka & Spark Streaming
Posted by Neelesh <ne...@gmail.com>.
Thanks Petr, Cody. This is a reasonable place to start for me. What I'm
trying to achieve
stream.foreachRDD {rdd=>
rdd.foreachPartition { p=>
Try(myFunc(...)) match {
case Sucess(s) => updatewatermark for this partition //of course,
expectation is that it will work only if there is a 1-1 mapping at this
point in time
case Failure() => Tell the driver not to generate a partition for
this kafka topic+partition for a while, by updating some shared state (zk)
}
}
}
I was looking for that mapping b/w kafka partition thats bound to a task
inside the task execution code, in cases where the intermediate operations
do not change partitions, shuffle etc.
-neelesh
On Fri, Sep 25, 2015 at 11:14 AM, Cody Koeninger <co...@koeninger.org> wrote:
>
> http://spark.apache.org/docs/latest/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers
>
> also has an example of how to close over the offset ranges so they are
> available on executors.
>
> On Fri, Sep 25, 2015 at 12:50 PM, Neelesh <ne...@gmail.com> wrote:
>
>> Hi,
>> We are using DirectKafkaInputDStream and store completed consumer
>> offsets in Kafka (0.8.2). However, some of our use case require that
>> offsets be not written if processing of a partition fails with certain
>> exceptions. This allows us to build various backoff strategies for that
>> partition, instead of either blindly committing consumer offsets regardless
>> of errors (because KafkaRDD as HasOffsetRanges is available only on the
>> driver) or relying on Spark's retry logic and continuing without remedial
>> action.
>>
>> I was playing with SparkListener and found that while one can listen on
>> taskCompletedEvent on the driver and even figure out that there was an
>> error, there is no way of mapping this task back to the partition and
>> retrieving offset range, topic & kafka partition # etc.
>>
>> Any pointers appreciated!
>>
>> Thanks!
>> -neelesh
>>
>
>
Re: Kafka & Spark Streaming
Posted by Cody Koeninger <co...@koeninger.org>.
http://spark.apache.org/docs/latest/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers
also has an example of how to close over the offset ranges so they are
available on executors.
On Fri, Sep 25, 2015 at 12:50 PM, Neelesh <ne...@gmail.com> wrote:
> Hi,
> We are using DirectKafkaInputDStream and store completed consumer
> offsets in Kafka (0.8.2). However, some of our use case require that
> offsets be not written if processing of a partition fails with certain
> exceptions. This allows us to build various backoff strategies for that
> partition, instead of either blindly committing consumer offsets regardless
> of errors (because KafkaRDD as HasOffsetRanges is available only on the
> driver) or relying on Spark's retry logic and continuing without remedial
> action.
>
> I was playing with SparkListener and found that while one can listen on
> taskCompletedEvent on the driver and even figure out that there was an
> error, there is no way of mapping this task back to the partition and
> retrieving offset range, topic & kafka partition # etc.
>
> Any pointers appreciated!
>
> Thanks!
> -neelesh
>