Objective

Within the context of a delta live table, I'm trying to merge two streams aggregation, but run into challenges. Is it possible to achieve such a join?

Context

Assume

- table trades stores a list of trades with their associated time stamps

- table trades_1d sums the value of all the trades on a given day

- table stock_price stores a given stock price at some (non-constant) sampling frequency

- table stock_price_1d averages the stock price over a given day

It would translate to something along those lines

import pyspark.sql.functions as sqlf
 
sdf_trades = spark.readStream.format("delta").table(f"stock_price")
sdf_price = spark.readStream.format("delta").table(f"stock_price")
 
 
w = sqlf.window("timetstamp", "24 hours")
 
sdf_trades_1d = (
    sdf_trades
    .groupby(w)
    .agg(sqlf.sum("trade_value"))
    .withColumn("window_end", sqlf.col("window.end"))
    .withColumn("window_start", sqlf.col("window.start"))
)
                
sdf_price_1d = (
    sdf_price
    .groupby(w)
    .agg(sqlf.avg("value"))
    .withColumn("window_end", sqlf.col("window.end"))
    .withColumn("window_start", sqlf.col("window.start"))
).withWatermark("window_end", "48 hours")
 
sdf = sdf_trades_1d.join(sdf_price_1d, "window_end", "left")

Issue

When running the pseudo code above, I get

"Append more error: Multiple streaming aggregations are not supported with stream DataFrames/Datasets"

Any suggestion on how I can make this work?