@Anonymous
You have to use writeStream instead of write.
dataDF.writeStream
.option("checkpointLocation", "path-for-checkpoint')
.foreachBatch { (batchDF: Dataset[Row], batchId: Long) =>
batchDF.write
.format("snowflake")
.options(sfOptions).option("dbtable", "demokinesisstream")
.mode(SaveMode.Append)
.save()
}
It is also recommended to use checkpoint directory to save the streaming metadata which help in restarting the pipeline from the previous point.
