How to stop spark structured streaming from listing all files in an S3 bucket every time

Question

I have a structured streaming job on pyspark which does some aggregations on a filesource. I have a kinesis firehose combines the data from an IoT type application and stores the data on an S3 location as a file per minute in different folders in the following folder structure -

s3://year/month/day/hour/

My spark structured streaming job seems to hat from listing all the files that are available in my S3 bucket. As the listing process seems to be taking more time than the processingTime that I've set. I get the following warning, I was wondering if there was a way to not let this happen.

18/06/15 14:28:35 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 60000 milliseconds, but spent 74364 milliseconds
18/06/15 14:28:42 WARN FileStreamSource: Listed 4449 file(s) in 6822.134244 ms
18/06/15 14:29:06 WARN FileStreamSource: Listed 4449 file(s) in 6478.381219 ms
18/06/15 14:30:08 WARN FileStreamSource: Listed 4450 file(s) in 8285.654031 ms

Answer 1

The S3 API List operation can only be used to retrieve all object keys in a bucket sharing a prefix. So it's simply impossible to list only new, unprocessed objects. The Databricks folks seem to have a solution where you set up S3 to create an SQS record when a new object is created. Spark then checks SQS for new objects and retrieves specific objects from S3 (i.e. no listing involved). Unfortunately this connector seems to be available only on Databricks clusters and hasn't been open sourced, so if you're using for example EMR, you can't use it (unless of course you implement the connector yourself).

Answer 2

A comment in the class FileStreamSource:

// Output a warning when listing files uses more than 2 seconds.

So, to get rid of this warning, you could reduce the amount of files processed every trigger:

maxFilesPerTrigger option can be set on the file source to ensure it takes < 2 seconds.

The first warning is the trigger interval you have set (60000) is shorter than the time taken (74364). Just increase the trigger interval to get rid of this.