Is it that in scala its allowed for derived class to have any return type ?
And streaming jar is originally created in scala so its allowed for
DirectKafkaInputDStream to return Option[KafkaRDD[K, V, U, T, R]]
compute method ?
On Tue, Aug 18, 2015 at 8:36 PM, Shushant Arora <shushantaror...@gmail.com>
wrote:
> looking at source code of
> org.apache.spark.streaming.kafka.DirectKafkaInputDStream
>
> override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {
> val untilOffsets = clamp(latestLeaderOffsets(maxRetries))
> val rdd = KafkaRDD[K, V, U, T, R](
> context.sparkContext, kafkaParams, currentOffsets, untilOffsets,
> messageHandler)
>
> currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)
> Some(rdd)
> }
>
>
> But in DStream its def compute (validTime: Time): Option[RDD[T]] ,
>
> So what should be the return type of custom DStream extends
> DirectKafkaInputDStream .
> Since I want the behaviour to be same as of DirectKafkaInputDStream in
> normal scenarios and return none in specific scenario.
>
> And why the same error did not come while extending
> DirectKafkaInputDStream from InputDStream ? Since new return type
> Option[KafkaRDD[K,
> V, U, T, R]] is not subclass of Option[RDD[T] so it should have been
> failed?
>
>
>
>
> On Tue, Aug 18, 2015 at 7:28 PM, Cody Koeninger <c...@koeninger.org>
> wrote:
>
>> The superclass method in DStream is defined as returning an Option[RDD[T]]
>>
>> On Tue, Aug 18, 2015 at 8:07 AM, Shushant Arora <
>> shushantaror...@gmail.com> wrote:
>>
>>> Getting compilation error while overriding compute method of
>>> DirectKafkaInputDStream.
>>>
>>>
>>> [ERROR] CustomDirectKafkaInputDstream.java:[51,83]
>>> compute(org.apache.spark.streaming.Time) in CustomDirectKafkaInputDstream
>>> cannot override compute(org.apache.spark.streaming.Time) in
>>> org.apache.spark.streaming.dstream.DStream; attempting to use incompatible
>>> return type
>>>
>>> [ERROR] found :
>>> scala.Option<org.apache.spark.streaming.kafka.KafkaRDD<byte[],byte[],kafka.serializer.DefaultDecoder,kafka.serializer.DefaultDecoder,byte[][]>>
>>>
>>> [ERROR] required: scala.Option<org.apache.spark.rdd.RDD<byte[][]>>
>>>
>>>
>>> class :
>>>
>>> public class CustomDirectKafkaInputDstream extends
>>> DirectKafkaInputDStream<byte[], byte[], kafka.serializer.DefaultDecoder,
>>> kafka.serializer.DefaultDecoder, byte[][]>{
>>>
>>> @Override
>>> public Option<KafkaRDD<byte[], byte[], DefaultDecoder, DefaultDecoder,
>>> byte[][]>> compute(
>>> Time validTime) {
>>>
>>> int processed=processedCounter.value();
>>> int failed = failedProcessingsCounter.value();
>>> if((processed==failed)){
>>> System.out.println("backing off since its 100 % failure");
>>> return Option.empty();
>>> }else{
>>> System.out.println("starting the stream ");
>>>
>>> return super.compute(validTime);
>>> }
>>> }
>>> }
>>>
>>>
>>> What should be the return type of compute method ? super class is
>>> returning Option<KafkaRDD<byte[], byte[], DefaultDecoder, DefaultDecoder,
>>> byte[][]>> but its expecting
>>> scala.Option<org.apache.spark.rdd.RDD<byte[][]>> from derived class . Is
>>> there something wring with code?
>>>
>>> On Mon, Aug 17, 2015 at 7:08 PM, Cody Koeninger <c...@koeninger.org>
>>> wrote:
>>>
>>>> Look at the definitions of the java-specific
>>>> KafkaUtils.createDirectStream methods (the ones that take a
>>>> JavaStreamingContext)
>>>>
>>>> On Mon, Aug 17, 2015 at 5:13 AM, Shushant Arora <
>>>> shushantaror...@gmail.com> wrote:
>>>>
>>>>> How to create classtag in java ?Also Constructor
>>>>> of DirectKafkaInputDStream takes Function1 not Function but
>>>>> kafkautils.createDirectStream allows function.
>>>>>
>>>>> I have below as overriden DirectKafkaInputDStream.
>>>>>
>>>>>
>>>>> public class CustomDirectKafkaInputDstream extends
>>>>> DirectKafkaInputDStream<byte[], byte[], kafka.serializer.DefaultDecoder,
>>>>> kafka.serializer.DefaultDecoder, byte[][]>{
>>>>>
>>>>> public CustomDirectKafkaInputDstream(
>>>>> StreamingContext ssc_,
>>>>> Map<String, String> kafkaParams,
>>>>> Map<TopicAndPartition, Object> fromOffsets,
>>>>> Function1<MessageAndMetadata<byte[], byte[]>, byte[][]> messageHandler,
>>>>> ClassTag<byte[]> evidence$1, ClassTag<byte[]> evidence$2,
>>>>> ClassTag<DefaultDecoder> evidence$3,
>>>>> ClassTag<DefaultDecoder> evidence$4, ClassTag<byte[][]> evidence$5) {
>>>>> super(ssc_, kafkaParams, fromOffsets, messageHandler, evidence$1,
>>>>> evidence$2,
>>>>> evidence$3, evidence$4, evidence$5);
>>>>> }
>>>>> @Override
>>>>> public Option<KafkaRDD<byte[], byte[], DefaultDecoder, DefaultDecoder,
>>>>> byte[][]>> compute(
>>>>> Time validTime) {
>>>>> int processe=processedCounter.value();
>>>>> int failed = failedProcessingsCounter.value();
>>>>> if((processed==failed)){
>>>>> System.out.println("backing off since its 100 % failure");
>>>>> return Option.empty();
>>>>> }else{
>>>>> System.out.println("starting the stream ");
>>>>>
>>>>> return super.compute(validTime);
>>>>> }
>>>>> }
>>>>>
>>>>>
>>>>>
>>>>> To create this stream
>>>>> I am using
>>>>> scala.collection.immutable.Map<String, String> scalakafkaParams =
>>>>> JavaConverters.mapAsScalaMapConverter(kafkaParams).asScala().toMap(Predef.<Tuple2<String,
>>>>> String>>conforms());
>>>>> scala.collection.immutable.Map<TopicAndPartition, Long>
>>>>> scalaktopicOffsetMap=
>>>>> JavaConverters.mapAsScalaMapConverter(topicOffsetMap).asScala().toMap(Predef.<Tuple2<TopicAndPartition,
>>>>> Long>>conforms());
>>>>>
>>>>> scala.Function1<MessageAndMetadata<byte[], byte[]>, byte[][]> handler
>>>>> = new Function<MessageAndMetadata<byte[], byte[]>, byte[][]>() {
>>>>> ..});
>>>>> JavaDStream<byte[][]> directKafkaStream = new
>>>>> CustomDirectKafkaInputDstream(jssc,scalakafkaParams ,scalaktopicOffsetMap,
>>>>> handler,byte[].class,byte[].class, kafka.serializer.DefaultDecoder.class,
>>>>> kafka.serializer.DefaultDecoder.class,byte[][].class);
>>>>>
>>>>>
>>>>>
>>>>> How to pass classTag to constructor in CustomDirectKafkaInputDstream ?
>>>>> And how to use Function instead of Function1 ?
>>>>>
>>>>>
>>>>>
>>>>> On Thu, Aug 13, 2015 at 12:16 AM, Cody Koeninger <c...@koeninger.org>
>>>>> wrote:
>>>>>
>>>>>> I'm not aware of an existing api per se, but you could create your
>>>>>> own subclass of the DStream that returns None for compute() under certain
>>>>>> conditions.
>>>>>>
>>>>>>
>>>>>>
>>>>>> On Wed, Aug 12, 2015 at 1:03 PM, Shushant Arora <
>>>>>> shushantaror...@gmail.com> wrote:
>>>>>>
>>>>>>> Hi Cody
>>>>>>>
>>>>>>> Can you help here if streaming 1.3 has any api for not consuming any
>>>>>>> message in next few runs?
>>>>>>>
>>>>>>> Thanks
>>>>>>>
>>>>>>> ---------- Forwarded message ----------
>>>>>>> From: Shushant Arora <shushantaror...@gmail.com>
>>>>>>> Date: Wed, Aug 12, 2015 at 11:23 PM
>>>>>>> Subject: spark streaming 1.3 doubts(force it to not consume anything)
>>>>>>> To: user <user@spark.apache.org>
>>>>>>>
>>>>>>>
>>>>>>> I Can't make my stream application batch interval to change at run
>>>>>>> time . Its always fixed and it always creates jobs at specified batch
>>>>>>> inetval and enqueue them if earleir batch is not finished.
>>>>>>>
>>>>>>> My requirement is to process the events and post them to some
>>>>>>> external server and if external server is down I want to increase the
>>>>>>> batch
>>>>>>> time - that is not possible but can I make it not to consume any
>>>>>>> messages
>>>>>>> in say next 5 successive runs ?
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
