dstream to hudi

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.sql.SaveMode._

val sc = spark.sparkContext
val ssc = new StreamingContext(sc, Seconds(1))

val inputPath = "/tmp/inputDir/"

val dStream = ssc.textFileStream(inputPath)

import org.apache.hudi.QuickstartUtils._
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._

val tableName = "hudi_trips_cow"
val dataGen = new DataGenerator

dStream.foreachRDD { rdd =>
  val batchDf = rdd.toDF()
  batchDf.write.format("hudi").
  options(getQuickstartWriteConfigs).
  option(PRECOMBINE_FIELD_OPT_KEY, "tpep_dropoff_datetime").
  option(RECORDKEY_FIELD_OPT_KEY, "tpep_pickup_datetime").
  option(PARTITIONPATH_FIELD_OPT_KEY, "VendorID").
  option(TABLE_NAME, tableName).
  mode(Append).
  save(basePath)
}

ssc.start()

dStream write to parquet

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.sql.SaveMode._

val sc = spark.sparkContext
val ssc = new StreamingContext(sc, Seconds(1))

val inputPath = "/tmp/inputDir/"

val dStream = ssc.textFileStream(inputPath)
val basePath = "file:///tmp/parquet"

dStream.foreachRDD { rdd =>
  val batchDf = rdd.toDF()
  batchDf.write.format("parquet").
  mode(Append).
  save(basePath)
}

ssc.start()

reading the parquet written data

val df = spark.read.format("parquet").load("/tmp/parquet/")
df: org.apache.spark.sql.DataFrame = [value: string]                            

scala> df.printSchema
root
 |-- value: string (nullable = true)


scala> df.show(2, false)
+-----+
|value|
+-----+
+-----+


scala> df.count
res2: Long = 0

Not sure the status of timeline. It seems that inputstream didn't generate proper data. Any log from hudi client side?