Spark Streaming

Structured Streaming reads are based on Hudi’s Incremental Query feature, therefore streaming read can return data for which commits and base files were not yet removed by the cleaner. You can control commits retention time.

Scala

  1. // spark-shell
  2. // reload data
  3. df.write.format("hudi").
  4.   options(getQuickstartWriteConfigs).
  5.   option("hoodie.datasource.write.precombine.field", "ts").
  6.   option("hoodie.datasource.write.recordkey.field", "uuid").
  7.   option("hoodie.datasource.write.partitionpath.field", "partitionpath").
  8.   option("hoodie.table.name", tableName).
  9.   mode(Overwrite).
  10.   save(basePath)
  12. // read stream and output results to console
  13. spark.readStream.
  14.   format("hudi").
  15.   load(basePath).
  16.   writeStream.
  17.   format("console").
  18.   start()
  20. // read stream to streaming df
  21. val df = spark.readStream.
  22.         format("hudi").
  23.         load(basePath)

Python

  1. # pyspark
  2. # reload data
  3. inserts = sc._jvm.org.apache.hudi.QuickstartUtils.convertToStringList(
  4.     dataGen.generateInserts(10))
  5. df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
  7. hudi_options = {
  8.     'hoodie.table.name': tableName,
  9.     'hoodie.datasource.write.recordkey.field': 'uuid',
  10.     'hoodie.datasource.write.partitionpath.field': 'partitionpath',
  11.     'hoodie.datasource.write.table.name': tableName,
  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. df.write.format("hudi"). \
  19.     options(**hudi_options). \
  20.     mode("overwrite"). \
  21.     save(basePath)
  23. # read stream to streaming df
  24. df = spark.readStream \
  25.     .format("hudi") \
  26.     .load(basePath)
  28. # ead stream and output results to console
  29. spark.readStream \
  30.     .format("hudi") \
  31.     .load(basePath) \
  32.     .writeStream \
  33.     .format("console") \
  34.     .start()

Spark SQL can be used within ForeachBatch sink to do INSERT, UPDATE, DELETE and MERGE INTO. Target table must exist before write.