Streaming Writes

Spark Streaming

You can write Hudi tables using spark’s structured streaming.

  • Scala
  • Python
  1. // spark-shell
  2. // prepare to stream write to new table
  3. import org.apache.spark.sql.streaming.Trigger
  5. val streamingTableName = "hudi_trips_cow_streaming"
  6. val baseStreamingPath = "file:///tmp/hudi_trips_cow_streaming"
  7. val checkpointLocation = "file:///tmp/checkpoints/hudi_trips_cow_streaming"
  9. // create streaming df
  10. val df = spark.readStream.
  11.         format("hudi").
  12.         load(basePath)
  14. // write stream to new hudi table
  15. df.writeStream.format("hudi").
  16.   options(getQuickstartWriteConfigs).
  17.   option("hoodie.datasource.write.precombine.field", "ts").
  18.   option("hoodie.datasource.write.recordkey.field", "uuid").
  19.   option("hoodie.datasource.write.partitionpath.field", "partitionpath").
  20.   option("hoodie.table.name", streamingTableName).
  21.   outputMode("append").
  22.   option("path", baseStreamingPath).
  23.   option("checkpointLocation", checkpointLocation).
  24.   trigger(Trigger.Once()).
  25.   start()
  1. # pyspark
  2. # prepare to stream write to new table
  3. streamingTableName = "hudi_trips_cow_streaming"
  4. baseStreamingPath = "file:///tmp/hudi_trips_cow_streaming"
  5. checkpointLocation = "file:///tmp/checkpoints/hudi_trips_cow_streaming"
  7. hudi_streaming_options = {
  8.     'hoodie.table.name': streamingTableName,
  9.     'hoodie.datasource.write.recordkey.field': 'uuid',
  10.     'hoodie.datasource.write.partitionpath.field': 'partitionpath',
  11.     'hoodie.datasource.write.table.name': streamingTableName,
  12.     'hoodie.datasource.write.operation': 'upsert',
  13.     'hoodie.datasource.write.precombine.field': 'ts',
  14.     'hoodie.upsert.shuffle.parallelism': 2,
  15.     'hoodie.insert.shuffle.parallelism': 2
  16. }
  18. # create streaming df
  19. df = spark.readStream 
  20.     .format("hudi") 
  21.     .load(basePath)
  23. # write stream to new hudi table
  24. df.writeStream.format("hudi") 
  25.     .options(**hudi_streaming_options) 
  26.     .outputMode("append") 
  27.     .option("path", baseStreamingPath) 
  28.     .option("checkpointLocation", checkpointLocation) 
  29.     .trigger(once=True) 
  30.     .start()