Spark Structured Streaming File Source Starting Offset

3
votes

Is there a way how to specify starting offset for Spark Structured File Stream Source ?

I am trying to stream parquets from HDFS:

spark.sql("SET spark.sql.streaming.schemaInference=true")

spark.readStream
  .parquet("/tmp/streaming/")
  .writeStream
  .option("checkpointLocation", "/tmp/streaming-test/checkpoint")
  .format("parquet")
  .option("path", "/tmp/parquet-sink")
  .trigger(Trigger.ProcessingTime(1.minutes))
  .start()

As I see, the first run is processing all available files detected in path, then save offsets to checkpoint location and process only new files, that is accept age and does not exist in files seen map.

I'm looking for a way, how to specify starting offset or timestamp or number of options to do not process all available files in the first run.

Is there a way I'm looking for?

2
apache-sparkapache-spark-sqlspark-streaming

2 Answers

7
votes

Thanks @jayfah, as far as I found, we might simulate Kafka 'latest' starting offsets using following trick:

  1. Run warn-up stream with option("latestFirst", true) and option("maxFilesPerTrigger", "1") with checkpoint, dummy sink and huge processing time. This way, warm-up stream will save latest file timestamp to checkpoint.

  2. Run real stream with option("maxFileAge", "0"), real sink using the same checkpoint location. In this case stream will process only newly available files.

Most probably that is not necessary for production and there is better way, e.g. reorganize data paths etc., but this way at least I found as answer for my question.

1
votes

The FileStreamSource has no option to specify a starting offset.

But you could set the option of latestFirst to true to ensure that it processes the latest files first (this option is false by default)

https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources

 spark.readStream
  .option("latestFirst", true)
  .parquet("/tmp/streaming/")
  .writeStream
  .option("checkpointLocation", "/tmp/streaming-test/checkpoint")
  .format("parquet")
  .option("path", "/tmp/parquet-sink")
  .trigger(Trigger.ProcessingTime(1.minutes))
  .start()