[Python] Left join in pyspark works like inner join

Here is my code:

joined = deals_parsed.alias("deals").join(
transactions_parsed.alias("transactions"),
expr("""
deals.symbol = transactions.symbol AND
deals.timestamp >= transactions.timestamp AND
deals.timestamp <= transactions.timestamp + interval 1 seconds
"""),
"leftOuter"
)

joined_windowed = (
joined
.withColumn("window", window(col("deals.timestamp"), "5 minutes"))
)

def process_batch(batch: DataFrame, _batch_id: int):
if batch.isEmpty():
logger.info(f"Batch {_batch_id} is empty.")
return
windows = batch.select("window").distinct().collect()
for window in windows:
window_start = window['window'].start
window_end = window['window'].end
window_df = batch.filter((batch.window.start == window_start) & (batch.window.end == window_end))
broker_count = window_df.filter(col("broker_id").isNotNull()).count()
exchange_count = window_df.count()
logger.info(f"Window: {window_start} - {window_end}")
logger.info(f"Broker Transactions: {broker_count}")
logger.info(f"Exchange Transactions: {exchange_count}")

query = (
joined_windowed
.writeStream
.outputMode("append")
.foreachBatch(process_batch)
.trigger(processingTime='5 minutes')
.start()
)

query.awaitTermination()


I have 2 streams. The number of elements in the deals stream is guaranteed to be smaller. However, after join, there are as many of them as transactions, because join works like inner and simply removes rows from deals without matches.

Please, help me find where my mistake.

Continue reading...