Empty folder in HDFS using Spark Streaming+Kafka

2
votes

I am using Spark Streaming + Kafka for data ingestion into HDFS. 

val ssc = new StreamingContext(sparkContext, Seconds(30))
val messageRecBased = KafkaUtils.createStream(ssc, zkQuorum, group, topic)
  .map(_._2)

Each 30 sec all data in Kafka queue will be stored in a separate folder in HDFS. Some folders contains a empty file called part-00000 because there is no data in the corresponding batch-interval (30 sec). I used the following lines to filter those folders:

messageRecBased.filter { x => x.size == 0 }
messageRecBased.repartition(1).saveAsTextFiles("PATH")

but it is not working and it still generates folders with empty file.

3
scalaapache-sparkhdfsspark-streaming

3 Answers

1
votes

If you look at the DStream.saveAsTextFiles() method definition, it simply spawns the RDD.saveAsObjectFile to each of the RDD in the DirectKafkaInputDStream.

  def saveAsObjectFiles(prefix: String, suffix: String = ""): Unit = ssc.withScope {
    val saveFunc = (rdd: RDD[T], time: Time) => {
      val file = rddToFileName(prefix, suffix, time)
      rdd.saveAsObjectFile(file)
    }
    this.foreachRDD(saveFunc)
  }

So instead of using DStream.saveAsTextFiles(), you can chose to write something of your own like:

messageRecBased.foreachRDD{ rdd =>
    rdd.repartition(1)
    if(!rdd.isEmpty)
        rdd.saveAsObjectFile("FILE_PATH")
}
0
votes

You can check if the partition is empty and if not then only save the RDD, like this. This code should prevent empty RDDs from saving up.

messageRecBased.partitions.isEmpty
0
votes

This is how i did to create new directory and avoid empty batches.

import java.time.format.DateTimeFormatter
import java.time.LocalDateTime

   messageRecBased.foreachRDD{ rdd =>
        rdd.repartition(1)
        val eachRdd = rdd.map(record => record.value)
        if(!eachRdd.isEmpty)
          eachRdd.saveAsTextFile("hdfs/location/"+DateTimeFormatter.ofPattern("yyyyMMddHHmmss").format(LocalDateTime.now)+"/")
      }