DLT fails with Queries with streaming sources must be executed with writeStream.start();

[Nastia]

Hi guys!

I am having an issue with passing the "streaming flow" between layers of the DLT.

first layer "ETD_Bz" is passing through, but then "ETD_Flattened_Bz" is failing with "pyspark.errors.exceptions.captured.AnalysisException: Queries with streaming sources must be executed with writeStream.start();" error.

Code:

# works successfully

@dlt.table( 

  name="ETD_Bz",

  temporary=False)

def Bronze():

    return (spark.readStream

                 .format("delta")

                 .option("skipChangeCommits", "true")

                 .table("default.tbl_raw_etd_data")

            )

 # function that flattens json

def process_raw_data(raw_tbl_name) :

    df = (spark.readStream # <- starts working once I am changing from readStream to read, but then it obviously stops processing incrementally

               .format("delta")

               .option("mergeSchema", "true")

               .table(raw_tbl_name)

         )

    json_schema = spark.read.json(df.rdd.map(lambda row: row.JsonString)).schema  # <- fails here

    kafka_df =  df.withColumn("JsonStruct", from_json(col("JsonString"), json_schema))

    fj = FlattenJson()

    kafka_flattened_json = fj.flatten_json(kafka_df)

    return kafka_flattened_json

# failing layer

@dlt.table(

  name="ETD_Flattened_Bz",

  spark_conf = {"spark.databricks.delta.schema.autoMerge.enabled" : "true"},

  temporary=False)

def Bronze_Flattend():

    return process_raw_data("live.ETD_Bz")

any help appreciated! Thank you very much in advance

[Nastia]

UPDATE: tried adding writeStream.start() like error suggested + as per other posts and ended up with following error/code:

@dlt.table( 

  name="ETD_Bz",

  temporary=False)

def Bronze():

    return (spark.readStream

                 .format("delta")

                 .option("skipChangeCommits", "true")

                 .table("default.tbl_raw_etd_data")

            )

 # function that flattens json

def process_raw_data(df, batchId) :

    json_schema = spark.read.json(df.rdd.map(lambda row: row.JsonString)).schema

    kafka_df =  df.withColumn("JsonStruct", from_json(col("JsonString"), json_schema))

    fj = FlattenJson()

    kafka_flattened_json = fj.flatten_json(kafka_df)

    return kafka_flattened_json

@dlt.table(

  name="ETD_Flattened_Bz",

  spark_conf = {"spark.databricks.delta.schema.autoMerge.enabled" : "true"},

  temporary=False)

def Bronze_Flattend():

  stream = (spark.readStream

            .format("delta")

            .option("skipChangeCommits", "true")

            .table("live.ETD_Bz")

            .writeStream

            .format("json")

            .outputMode("append")

            .foreachBatch(process_raw_data)

            .table("default.tbl_bz_tmp_etd_data")

            .start()

            .awaitTermination())

  return (spark.readStream

                 .format("delta")

                 .option("skipChangeCommits", "true")

                 .table("default.tbl_bz_tmp_etd_data")

            )

getting following error:

"py4j.protocol.Py4JJavaError: An error occurred while calling o687.toTable. : org.apache.spark.SparkClassNotFoundException: [DATA_SOURCE_NOT_FOUND] Failed to find data source: foreachBatch. Please find packages at `https://spark.apache.org/third-party-projects.html`."

[Kaniz_Fatma]

Hi @Nastia, Thank you for sharing the updated code! The error message you’re seeing, “org.apache.spark.SparkClassNotFoundException: [DATA_SOURCE_NOT_FOUND] Failed to find data source: foreachBatch,” indicates that the foreachBatch operation is not recognized by Databricks Live Tables (DLT). Unfortunately, DLT does not directly support foreachBatch for streaming queries.

• Instead of directly writing to the target table within the foreachBatch operation, create a temporary table to hold the intermediate results.
• After processing each micro-batch, write the results to the temporary table.
• Finally, use a separate job or process to periodically merge the data from the temporary table into your target table.

 I hope this helps! If you have any further questions or need additional assistance, feel free to ask. 😊