cancel
Showing results for 
Search instead for 
Did you mean: 
Data Engineering
Join discussions on data engineering best practices, architectures, and optimization strategies within the Databricks Community. Exchange insights and solutions with fellow data engineers.
cancel
Showing results for 
Search instead for 
Did you mean: 

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

Nastia
New Contributor III

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
2 REPLIES 2

Nastia
New Contributor III

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`."

Hi @NastiaThank 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. 😊
 
Join 100K+ Data Experts: Register Now & Grow with Us!

Excited to expand your horizons with us? Click here to Register and begin your journey to success!

Already a member? Login and join your local regional user group! If there isn’t one near you, fill out this form and we’ll create one for you to join!