Thanks @Dan Burkert - I've done some more work on this and it seems that its possible, but I ran into a few issues with it. There is no sink for kudu in spark streaming, so you have to do a foreachwriter sink instead. Then convert the state class to a dataframe and upsert each row at a time. I backed out of this approach as when doing a Seq() on the value output by the foreachwriter to convert to a dataframe, the spark session was lost within the process stage, and I was unable to initiate spark session at this area. My code incase anyone has tried this before and knows a way around it: df.select([...]) .as[IoTState] .groupbyKey(_.assetId) .mapGroupsWithState(GroupStateTimeout.NoTimeout)(updateIoTStateFromEvent) .writeStream .option("checkpointLocation",conf.stateCheckPoint) .format("console") .foreach(new ForeachWriter[IoTState] { override def open(partitionId: Long, version: Long): Boolean = { true } override def process(value: IoTState): Unit = { val valueDF: DataFrame = Seq(value).toDF() kuduContext.upsertRows(valueDF, conf.kuduTable) } override def close(errorOrNull: Throwable): Unit = { } }) .outputMode("update") .trigger(Trigger.ProcessingTime("2 seconds")) .start() .awaitTermination()
... View more
