spark.scala

val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](job.ssc, kafkaParams, topics)

stream.foreachRDD(rdd => 
  import sparkSession.implicits._
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  val df = rdd.toDF()                
  df.write.format("sequence").mode("append").save(s"$workDir/$targetFile")
  stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)                 
})