You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Dan Dong <do...@gmail.com> on 2017/09/06 12:38:38 UTC
Multiple Kafka topics processing in Spark 2.2
Hi, All,
I have one issue here about how to process multiple Kafka topics in a
Spark 2.* program. My question is: How to get the topic name from a message
received from Kafka? E.g:
......
val messages = KafkaUtils.createDirectStream[String, String,
StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSet)
// Get the lines, split them into words, count the words and print
val lines = messages.map(_._2)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
wordCounts.print()
......
Kafka send the messages in multiple topics through console producer for
example. But when Spark receive the message, how it will know which topic
is this piece of message coming from? Thanks a lot for any of your helps!
Cheers,
Dan
Re: Multiple Kafka topics processing in Spark 2.2
Posted by kant kodali <ka...@gmail.com>.
@Dan shouldn't you be using Dataset/Dataframes ? I heard it is recommended
to use Dataset and Dataframes than using Dstreams since Dstreams is in
maintenance mode.
On Mon, Sep 11, 2017 at 7:41 AM, Cody Koeninger <co...@koeninger.org> wrote:
> If you want an "easy" but not particularly performant way to do it, each
> org.apache.kafka.clients.consumer.ConsumerRecord has a topic.
>
> The topic is going to be the same for the entire partition as long as you
> haven't shuffled, hence the examples on how to deal with it at a partition
> level.
>
> On Fri, Sep 8, 2017 at 8:29 PM, Dan Dong <do...@gmail.com> wrote:
>
>> Hi,Alonso.
>> Thanks! I've read about this but did not quite understand it. To pick
>> out the topic name of a kafka message seems a simple task but the example
>> code looks so complicated with redundent info. Why do we need offsetRanges
>> here and do we have a easy way to achieve this?
>>
>> Cheers,
>> Dan
>>
>>
>> 2017-09-06 21:17 GMT+08:00 Alonso Isidoro Roman <al...@gmail.com>:
>>
>>> Hi, reading the official doc
>>> <http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html>,
>>> i think you can do it this way:
>>>
>>> import org.apache.spark.streaming.kafka._
>>>
>>> val directKafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
>>>
>>> ssc, kafkaParams, topicsSet)
>>>
>>>
>>> // Hold a reference to the current offset ranges, so it can be used downstream
>>> var offsetRanges = Array.empty[OffsetRange]
>>>
>>> directKafkaStream.transform { rdd =>
>>> offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
>>> rdd
>>> }.map {
>>> ...
>>> }.foreachRDD { rdd =>
>>> for (o <- offsetRanges) {
>>> println(*s"${o.topic}* ${o.partition} ${o.fromOffset} ${o.untilOffset}")
>>> }
>>>
>>> }
>>>
>>>
>>> 2017-09-06 14:38 GMT+02:00 Dan Dong <do...@gmail.com>:
>>>
>>>> Hi, All,
>>>> I have one issue here about how to process multiple Kafka topics in a
>>>> Spark 2.* program. My question is: How to get the topic name from a message
>>>> received from Kafka? E.g:
>>>>
>>>> ......
>>>> val messages = KafkaUtils.createDirectStream[String, String,
>>>> StringDecoder, StringDecoder](
>>>> ssc, kafkaParams, topicsSet)
>>>>
>>>> // Get the lines, split them into words, count the words and print
>>>> val lines = messages.map(_._2)
>>>> val words = lines.flatMap(_.split(" "))
>>>> val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
>>>> wordCounts.print()
>>>> ......
>>>>
>>>> Kafka send the messages in multiple topics through console producer for
>>>> example. But when Spark receive the message, how it will know which topic
>>>> is this piece of message coming from? Thanks a lot for any of your helps!
>>>>
>>>> Cheers,
>>>> Dan
>>>>
>>>
>>>
>>>
>>> --
>>> Alonso Isidoro Roman
>>> [image: https://]about.me/alonso.isidoro.roman
>>>
>>> <https://about.me/alonso.isidoro.roman?promo=email_sig&utm_source=email_sig&utm_medium=email_sig&utm_campaign=external_links>
>>>
>>
>>
>
Re: Multiple Kafka topics processing in Spark 2.2
Posted by Cody Koeninger <co...@koeninger.org>.
If you want an "easy" but not particularly performant way to do it,
each org.apache.kafka.clients.consumer.ConsumerRecord
has a topic.
The topic is going to be the same for the entire partition as long as you
haven't shuffled, hence the examples on how to deal with it at a partition
level.
On Fri, Sep 8, 2017 at 8:29 PM, Dan Dong <do...@gmail.com> wrote:
> Hi,Alonso.
> Thanks! I've read about this but did not quite understand it. To pick
> out the topic name of a kafka message seems a simple task but the example
> code looks so complicated with redundent info. Why do we need offsetRanges
> here and do we have a easy way to achieve this?
>
> Cheers,
> Dan
>
>
> 2017-09-06 21:17 GMT+08:00 Alonso Isidoro Roman <al...@gmail.com>:
>
>> Hi, reading the official doc
>> <http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html>,
>> i think you can do it this way:
>>
>> import org.apache.spark.streaming.kafka._
>>
>> val directKafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
>>
>> ssc, kafkaParams, topicsSet)
>>
>>
>> // Hold a reference to the current offset ranges, so it can be used downstream
>> var offsetRanges = Array.empty[OffsetRange]
>>
>> directKafkaStream.transform { rdd =>
>> offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
>> rdd
>> }.map {
>> ...
>> }.foreachRDD { rdd =>
>> for (o <- offsetRanges) {
>> println(*s"${o.topic}* ${o.partition} ${o.fromOffset} ${o.untilOffset}")
>> }
>>
>> }
>>
>>
>> 2017-09-06 14:38 GMT+02:00 Dan Dong <do...@gmail.com>:
>>
>>> Hi, All,
>>> I have one issue here about how to process multiple Kafka topics in a
>>> Spark 2.* program. My question is: How to get the topic name from a message
>>> received from Kafka? E.g:
>>>
>>> ......
>>> val messages = KafkaUtils.createDirectStream[String, String,
>>> StringDecoder, StringDecoder](
>>> ssc, kafkaParams, topicsSet)
>>>
>>> // Get the lines, split them into words, count the words and print
>>> val lines = messages.map(_._2)
>>> val words = lines.flatMap(_.split(" "))
>>> val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
>>> wordCounts.print()
>>> ......
>>>
>>> Kafka send the messages in multiple topics through console producer for
>>> example. But when Spark receive the message, how it will know which topic
>>> is this piece of message coming from? Thanks a lot for any of your helps!
>>>
>>> Cheers,
>>> Dan
>>>
>>
>>
>>
>> --
>> Alonso Isidoro Roman
>> [image: https://]about.me/alonso.isidoro.roman
>>
>> <https://about.me/alonso.isidoro.roman?promo=email_sig&utm_source=email_sig&utm_medium=email_sig&utm_campaign=external_links>
>>
>
>
Re: Multiple Kafka topics processing in Spark 2.2
Posted by Dan Dong <do...@gmail.com>.
Hi,Alonso.
Thanks! I've read about this but did not quite understand it. To pick out
the topic name of a kafka message seems a simple task but the example code
looks so complicated with redundent info. Why do we need offsetRanges here
and do we have a easy way to achieve this?
Cheers,
Dan
2017-09-06 21:17 GMT+08:00 Alonso Isidoro Roman <al...@gmail.com>:
> Hi, reading the official doc
> <http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html>,
> i think you can do it this way:
>
> import org.apache.spark.streaming.kafka._
>
> val directKafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
>
> ssc, kafkaParams, topicsSet)
>
>
> // Hold a reference to the current offset ranges, so it can be used downstream
> var offsetRanges = Array.empty[OffsetRange]
>
> directKafkaStream.transform { rdd =>
> offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
> rdd
> }.map {
> ...
> }.foreachRDD { rdd =>
> for (o <- offsetRanges) {
> println(*s"${o.topic}* ${o.partition} ${o.fromOffset} ${o.untilOffset}")
> }
>
> }
>
>
> 2017-09-06 14:38 GMT+02:00 Dan Dong <do...@gmail.com>:
>
>> Hi, All,
>> I have one issue here about how to process multiple Kafka topics in a
>> Spark 2.* program. My question is: How to get the topic name from a message
>> received from Kafka? E.g:
>>
>> ......
>> val messages = KafkaUtils.createDirectStream[String, String,
>> StringDecoder, StringDecoder](
>> ssc, kafkaParams, topicsSet)
>>
>> // Get the lines, split them into words, count the words and print
>> val lines = messages.map(_._2)
>> val words = lines.flatMap(_.split(" "))
>> val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
>> wordCounts.print()
>> ......
>>
>> Kafka send the messages in multiple topics through console producer for
>> example. But when Spark receive the message, how it will know which topic
>> is this piece of message coming from? Thanks a lot for any of your helps!
>>
>> Cheers,
>> Dan
>>
>
>
>
> --
> Alonso Isidoro Roman
> [image: https://]about.me/alonso.isidoro.roman
>
> <https://about.me/alonso.isidoro.roman?promo=email_sig&utm_source=email_sig&utm_medium=email_sig&utm_campaign=external_links>
>
Re: Multiple Kafka topics processing in Spark 2.2
Posted by Alonso Isidoro Roman <al...@gmail.com>.
Hi, reading the official doc
<http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html>,
i think you can do it this way:
import org.apache.spark.streaming.kafka._
val directKafkaStream = KafkaUtils.createDirectStream[String,
String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSet)
// Hold a reference to the current offset ranges, so it can be used downstream
var offsetRanges = Array.empty[OffsetRange]
directKafkaStream.transform { rdd =>
offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd
}.map {
...
}.foreachRDD { rdd =>
for (o <- offsetRanges) {
println(*s"${o.topic}* ${o.partition} ${o.fromOffset} ${o.untilOffset}")
}
}
2017-09-06 14:38 GMT+02:00 Dan Dong <do...@gmail.com>:
> Hi, All,
> I have one issue here about how to process multiple Kafka topics in a
> Spark 2.* program. My question is: How to get the topic name from a message
> received from Kafka? E.g:
>
> ......
> val messages = KafkaUtils.createDirectStream[String, String,
> StringDecoder, StringDecoder](
> ssc, kafkaParams, topicsSet)
>
> // Get the lines, split them into words, count the words and print
> val lines = messages.map(_._2)
> val words = lines.flatMap(_.split(" "))
> val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
> wordCounts.print()
> ......
>
> Kafka send the messages in multiple topics through console producer for
> example. But when Spark receive the message, how it will know which topic
> is this piece of message coming from? Thanks a lot for any of your helps!
>
> Cheers,
> Dan
>
--
Alonso Isidoro Roman
[image: https://]about.me/alonso.isidoro.roman
<https://about.me/alonso.isidoro.roman?promo=email_sig&utm_source=email_sig&utm_medium=email_sig&utm_campaign=external_links>