Regarding : How to use Row_number() in dlt pipelines

anusha98
New Contributor II

‎10-20-2025 10:22 AM

We have two streaming tables : customer_info and customer_info_history and we  joined them using full join to create temp table in pyspark and now we want to eliminate the de-duped records from this temp table. Tried using row_number() but facing below error ..if anyone have solution to fix this error, kindly let me know.

error:Failed to start stream table in either append mode or complete mode. Append mode error: [NON_TIME_WINDOW_NOT_SUPPORTED_IN_STREAMING] Window function is not supported in ROW_NUMBER() (as column `row_num`) on streaming DataFrames/Datasets. Structured Streaming only supports time-window aggregation using the WINDOW function.

code:

@dlt.table(name="table")
def temp_latest_email():
    df = dlt.read_stream("table")

    window_spec = Window.partitionBy("col1", "col2", "col3").orderBy(col("col3").desc())

    df_with_rownum = df.withColumn("row_num", row_number().over(window_spec))

    return df_with_rownum.filter(col("row_num") == 1).drop("row_num")