How to access file paths in records from Kafka and create Datasets from?

user_3547993 0
Viewed 548

I am using Java.

I am receiving a filepath over Kafka messages. And I need to load this file into a spark RDD, process it, and dump it into HDFS.

I am able to retrieve the filepath from Kafka message. And I wish to create a dataset / RDD over this file.

I cannot run a map function on Kafka message dataset. It errors out with a NPE as sparkContext is not available on worker.

I cannot run a foreach on the Kafka messages dataset. It errors out with message:

Queries with streaming sources must be executed with writeStream.start();"

I cannot collect the data received from kafka message dataset, as it errors out with message

Queries with streaming sources must be executed with writeStream.start();;

I guess this must be a very general use-case and must be running in lot of setups.

How can I load the file as RDD from the paths that I receive in Kafka message?

SparkSession spark = SparkSession.builder()
.appName("MyKafkaStreamReader")
    .master("local[4]")
.config("spark.executor.memory", "2g")
.getOrCreate();

// Create DataSet representing the stream of input lines from kafka
Dataset<String> kafkaValues = spark.readStream()
.format("kafka")
    .option("spark.streaming.receiver.writeAheadLog.enable", true)
    .option("kafka.bootstrap.servers", Configuration.KAFKA_BROKER)
    .option("subscribe", Configuration.KAFKA_TOPIC)
    .option("fetchOffset.retryIntervalMs", 100)
    .option("checkpointLocation", "file:///tmp/checkpoint")
.load()
    .selectExpr("CAST(value AS STRING)").as(Encoders.STRING());

Dataset<String> messages = kafkaValues.map(x -> {
  ObjectMapper mapper = new ObjectMapper();
  String m = mapper.readValue(x.getBytes(), String.class);
  return m;
}, Encoders.STRING() );

// ====================
// TEST 1 : FAILS
// ====================    
// CODE TRYING TO execute MAP on the received RDD 
// This fails with a Null pointer exception because "spark" is not available on worker node

/*
Dataset<String> statusRDD = messages.map(message -> {

  // BELOW STATEMENT FAILS
  Dataset<Row> fileDataset = spark.read().option("header", "true").csv(message); 
  Dataset<Row> dedupedFileDataset = fileDataset.dropDuplicates();
  dedupedFileDataset.rdd().saveAsTextFile(getHdfsLocation());
  return getHdfsLocation();

}, Encoders.STRING());

  StreamingQuery query2 = statusRDD.writeStream().outputMode("append").format("console").start();
  */

// ====================    
// TEST 2 : FAILS
// ====================    
// CODE BELOW FAILS WITH EXCEPTION 
// "Queries with streaming sources must be executed with writeStream.start();;"
// Hence, processing the deduplication on the worker side using
/*
JavaRDD<String> messageRDD = messages.toJavaRDD();

messageRDD.foreach( message -> {

  Dataset<Row> fileDataset = spark.read().option("header", "true").csv(message);
  Dataset<Row> dedupedFileDataset = fileDataset.dropDuplicates();
  dedupedFileDataset.rdd().saveAsTextFile(getHdfsLocation());

});
*/

// ====================    
// TEST 3 : FAILS
// ====================
// CODE TRYING TO COLLECT ALSO FAILS WITH EXCEPTION
// "Queries with streaming sources must be executed with writeStream.start();;"
// List<String> mess = messages.collectAsList();

Any idea on how can I read create the file-paths and create RDDs over the files?

spark-structured-streaming apache-kafka apache-spark java
1 Answers

In Structured Streaming, I don't think that there's a way to reify the data in one stream to be used as a parameter for a Dataset operation.

In the Spark ecosystem, this is possible by combining Spark Streaming and Spark SQL (Datasets). We can use Spark Streaming to consume the Kafka topic and then, using Spark SQL, we can load the corresponding data and apply the desired process.

Such a job would look roughly like this: (This is in Scala, Java code will follow the same structure. Only that the actual code is a bit more verbose)

// configure and create spark Session

val spark = SparkSession
    .builder
    .config(...)
    .getOrCreate()

// create streaming context with a 30-second interval - adjust as required
val streamingContext = new StreamingContext(spark.sparkContext, Seconds(30))

// this uses Kafka080 client. Kafka010 has some subscription differences

val kafkaParams = Map[String, String](
  "metadata.broker.list" -> kafkaBootstrapServer,
  "group.id" -> "job-group-id",
  "auto.offset.reset" -> "largest",
  "enable.auto.commit" -> (false: java.lang.Boolean).toString
)

// create a kafka direct stream
val topics = Set("topic")
val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
     streamingContext, kafkaParams, topics)

// extract the values from the kafka message
val dataStream = stream.map{case (id, data) => data}     

// process the data
dataStream.foreachRDD { dataRDD => 
  // get all data received in the current interval
  // We are assuming that this data fits in memory. 
  // We're not processing a million files per second, are we?
  val files = dataRDD.collect()
  files.foreach{ file => 
    // this is the process proposed in the question --
    // notice how we have access to the spark session in the context of the foreachRDD
    val fileDataset = spark.read().option("header", "true").csv(file) 
    val dedupedFileDataset = fileDataset.dropDuplicates()
    // this can probably be written in terms of the dataset api
    //dedupedFileDataset.rdd().saveAsTextFile(getHdfsLocation());
    dedupedFileDataset.write.format("text").mode("overwrite").save(getHdfsLocation())
  }
}

// start the streaming process
streamingContext.start()
streamingContext.awaitTermination()
user_7640400
Related