[Python] Spark Structured Streaming, empty result for a stream-stream inner join to compute...

I've been trying to use Spark Structured Streaming for a hypothetical data streaming question against a non-commercial IMDB dataset.

The original raw data: https://datasets.imdbws.com/

The original schema: https://zindilis.com/posts/imdb-non-commercial-datasets-schema/

The question:

"Determine the ranking for titles, with the ranking determined by: (numVotes/averageNumVotes) * averageRating"

Where the ratings are assumed to be continuously streamed in. As such I amended the schema to make it an event log, with an additional timestamp field.

I decided to use Spark Structured Streaming because I was already familiar with batching in Spark, and their documentation here https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html gave me the impression that the state part of my problem could be managed by the framework without any custom state handling. At least with the latest version, anyway. Essentially it's a streaming aggregation followed by a streaming inner join.

However I ran into a problem: I am getting empty result from the join, even though both join inputs are populated with rows where the join keys match when I compute those separately (same code, except for watermark handling and sinks). Also, the same code produces the expected join output when run in a non-streaming manner.

Here are some notebooks illustrating each of the above cases:

1. 
   Streaming inner join with unexpected empty result: https://github.com/alza-bitz/plaype...y-join/imdb/spark_streaming_join_result.ipynb
   
   
2. 
   Streaming without join but with expected join inputs: https://github.com/alza-bitz/plaype...y-join/imdb/spark_streaming_join_inputs.ipynb
   
   
3. 
   Non-streaming equivalent for reference: https://github.com/alza-bitz/playpen-python/blob/spark-streaming-empty-join/imdb/spark_sql.ipynb
   

These should also be runnable directly in VSCode Dev Containers, the project includes a suitable devcontainer.json that installs Spark version 3.5.3.

I'm posting here to get some opinions and feedback before I raise a bug on the Spark Jira. At this stage, I'm unclear on whether this is a bug or whether I have misinterpreted the docs.

I did find a reference to what might be the same issue on Reddit, with a suggestion of updating to version 3.5.4: https://www.reddit.com/r/apachespark/comments/1cb056c/structured_streaming_join_on_window_column/

I tried that, but unfortunately the problem remained.

Thanks!

Continue reading...