spark streaming from kafka how to specify cut off time for events polled

user_1470234 0
Viewed 999

I have spark streaming application which runs End of day and consumes kafka events send by the upstream application.Currently upstream application keeps pushing new data the whole day and my consumer ends up consuming it. I want to limit the events consumed based on a cutoff say 6pm daily.Is there a way to specify a cut off to limit the events consumed based on a cutoff say kafka event timestamp or something .Below is the consumer code 

  KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topicSet, kafkaParams))
spark-streaming kafka-consumer-api apache-kafka apache-spark scala
3 Answers

You can just filter out the events during processing on the basis of timeStamp or time or any field. For example lets suppose your event is JSON and it have field called hour which is event time hour value. You can easily choose only event which got created before 6 like below.

directStream.foreachRDD { rdd =>
        val eventDfRDD = rdd.filter(record => {
          val option = JSON.parseFull(record).get.asInstanceOf[Map[String, String]]
          option.get("hour") < 1800
        })
      }
user_26215070

When you are declaring the streaming context, we can mention the cut off time to create dsstream and we can pass that value to createDirectStream parameter. Please find the code snap. In below code , 5 secs as cut-off time . so every 5 sec DStream RDD will be created.

sc = spark.sparkContext
ssc = StreamingContext(sc,5)
kvs = KafkaUtils.createDirectStream(ssc, ['Topic-name'], {"metadata.broker.list": 'Server-name:port-number'},valueDecoder=serializer.decode_message)
user_95990910

This is the solution i implemented

1: Store current time in a variable when the spark streaming job starts

val cuttoffTime =System.currentTimeMillis()

2: Create DirectStream

val directKafkaStream=   KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topicSet, kafkaParams))

3: Apply the filter criteria In the foreach loop apply the filter criteria like below

directKafkaStream.foreachRDD { rdd => val filteredRdd = rdd.filter(_.timestamp() < cuttoffTime )

user_14702340
Related