You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Guillermo Ortiz <ko...@gmail.com> on 2016/03/10 17:56:11 UTC
Problem with union of DirectStream
I have a DirectStream and process data from Kafka,
val directKafkaStream = KafkaUtils.createDirectStream[String, String,
StringDecoder, StringDecoder](ssc, kafkaParams1, topics1.toSet)
directKafkaStream.foreachRDD { rdd =>
val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
When I have added a new DirectStream and do a union between both it doesn't
work. I thought that it was the same type, but I got a ClassCastException
val directKafkaStream = KafkaUtils.createDirectStream[String, String,
StringDecoder, StringDecoder](ssc, kafkaParams1, topics1.toSet)
val directKafkaStream2 = KafkaUtils.createDirectStream[String, String,
StringDecoder, StringDecoder](ssc, kafkaParams2, topics2.toSet)
val kafkaStream = directKafkaStream.union(directKafkaStream2)
kafkaStream.foreachRDD { rdd =>
val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
-->Exception
Exception in thread "main" java.lang.ClassCastException:
org.apache.spark.rdd.UnionRDD cannot be cast to
org.apache.spark.streaming.kafka.HasOffsetRanges
at
com.produban.metrics.MetricsSpark$$anonfun$main$1.apply(MetricsSpark.scala:72)
I guessed that rdd.union(rdd2) gives same type of RDD..
Re: Problem with union of DirectStream
Posted by Cody Koeninger <co...@koeninger.org>.
If you do any RDD transformation, it's going to return a different RDD
than the original.
The implication for casting to HasOffsetRanges is specifically called
out in the docs at
http://spark.apache.org/docs/latest/streaming-kafka-integration.html#approach-2-direct-approach-no-receivers
On Thu, Mar 10, 2016 at 10:56 AM, Guillermo Ortiz <ko...@gmail.com> wrote:
> I have a DirectStream and process data from Kafka,
>
> val directKafkaStream = KafkaUtils.createDirectStream[String, String,
> StringDecoder, StringDecoder](ssc, kafkaParams1, topics1.toSet)
> directKafkaStream.foreachRDD { rdd =>
> val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
>
> When I have added a new DirectStream and do a union between both it doesn't
> work. I thought that it was the same type, but I got a ClassCastException
>
>
> val directKafkaStream = KafkaUtils.createDirectStream[String, String,
> StringDecoder, StringDecoder](ssc, kafkaParams1, topics1.toSet)
> val directKafkaStream2 = KafkaUtils.createDirectStream[String, String,
> StringDecoder, StringDecoder](ssc, kafkaParams2, topics2.toSet)
> val kafkaStream = directKafkaStream.union(directKafkaStream2)
>
> kafkaStream.foreachRDD { rdd =>
> val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
> -->Exception
>
> Exception in thread "main" java.lang.ClassCastException:
> org.apache.spark.rdd.UnionRDD cannot be cast to
> org.apache.spark.streaming.kafka.HasOffsetRanges
> at
> com.produban.metrics.MetricsSpark$$anonfun$main$1.apply(MetricsSpark.scala:72)
>
> I guessed that rdd.union(rdd2) gives same type of RDD..
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org