structured streaming read based on kafka partitions

Question

I am using spark structured Streaming to Read incoming messages from a Kafka topic and write to multiple parquet tables based on the incoming message So i created a single readStream as Kafka source is common and for each parquet table created separate write stream in a loop . This works fine but the readstream is creating a bottleneck as for each writeStream it create a readStream and there is no way to cache the dataframe which is already read.

val kafkaDf=spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", conf.servers)
      .option("subscribe", conf.topics)
      //  .option("earliestOffset","true")
      .option("failOnDataLoss",false)
      .load()

foreach table   {  
//filter the data from source based on table name
//write to parquet
 parquetDf.writeStream.format("parquet")
        .option("path", outputFolder + File.separator+ tableName)
        .option("checkpointLocation", "checkpoint_"+tableName)
        .outputMode("append")
        .trigger(Trigger.Once())
       .start()
}

Now every write stream is creating a new consumer group and reading entire data from Kafka and then doing the filter and writing to Parquet. This is creating a huge overhead. To avoid this overhead, I can partition the Kafka topic to have as many partitions as the number of tables and then the readstream should only read from a given partition. But I don't see a way to specify partition details as part of Kafka read stream.

You shouldn't need to manually assign Spark tasks to Kafka partitions. You would only need to scale up the number of executors to the number of partitions. — OneCricketeer
Thanks Cricket. so basically while writing to kafka i m using the tablename as key . so in the case of 1 readStream to many writeStream . I see that every writer is reading from kafka again so is there a way to limit it reading only the message it is going to write? say i hv table1 and table2 and while writing i put the kafka key as table1 and table2 . Now is there a way for writestream for table1 to read only message for table1 and not to read message for table2. i hv around 50 table so this is creating a huge bottleneck — Ajith Kannan
@Ajith .. have you got any solution for this? i am looking for same — BigD

dunlu_98k dunlu_98k · Accepted Answer · 2018-10-14T09:21:31

if data volume is not that high, write your own sink, collect data of each micro-batch , then you should be able to cache that dataframe and write to different locations, need some tweaks though but it will work

structured streaming read based on kafka partitions

1 Answers