Question: Delta Live Table, multiple streaming sources to the single target

bzh
New Contributor

‎08-14-2023 06:27 PM

We are trying to writing multiple sources to the same target table using DLT, but getting the below errors. 

Not sure what we are missing here in the code....

File /databricks/spark/python/dlt/api.py:817, in apply_changes(target, source, keys, sequence_by, where, ignore_null_updates, apply_as_deletes, apply_as_truncates, column_list, except_column_list, stored_as_scd_type, track_history_column_list, track_history_except_column_list, flow_name) 814 raise RuntimeError("Only SCD Type 1 and SCD Type 2 are supported for now.")

 

dlt.create_streaming_table(name='unified_events_test_11')

dlt.apply_changes(
target="unified_events_test_11",
source="unified_events_pv_raw",
keys=["event_id"],
sequence_by=F.col("cdcTimestamp"),
apply_as_deletes=F.expr("operation = 'D'"),
except_column_list=["operation", "cdcTimestamp"],
stored_as_scd_type=1
)
 
dlt.apply_changes(
target="unified_events_test_11",
source="unified_events_wc_raw",
keys=["event_id"],
sequence_by=F.col("cdcTimestamp"),
apply_as_deletes=F.expr("operation = 'D'"),
except_column_list=["operation", "cdcTimestamp"],
stored_as_scd_type=1
)
 
Note: unified_events_pv_raw and unified_events_wc_raw are streaming table
 
@dlt.table(
name="unified_events_wc_raw"
)
def unified_events_wc_raw():
df = (
spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("sep", "||||")
.schema(wallet_connect_schema)
.load("dbfs:/FileStore/wallet_connect_events")
)
0 Kudos

SparkJun
Databricks Employee
Databricks Employee

‎09-06-2023 04:38 PM

Hi bzh, I understand that you are writing multiple sources to the same target table on DLT and you are following the SCD type 1 process as documented (https://docs.databricks.com/en/delta-live-tables/cdc.html#process-scd-type-1-updates). However, the way how you declared two dlt.apply_changes from two different sources triggered this error. Firstly, you can narrow down the issue to troubleshoot by removing the 1st or the 2nd apply_changes declaration, so that you can test it with only one declaration and see if the same error persists. If that doesn't throw you an error, then try to change your code to declare one apply_changes function with two sources. The apply_changes function allows you to specify multiple sources by providing a list of tables or views as the source parameter. Please refer to this sample code:

import dlt

@dlt.table
def source1():
    return dlt.read("table1")

@dlt.table
def source2():
    return dlt.read("table2")

@dlt.table
def target():
    return dlt.apply_changes(
        target="target_table",
        source=["source1", "source2"],
        keys=["key_column"]
    )

Hope this helps, and happy coding on DLT, bzh!

0 Kudos

jose_gonzalez
Databricks Employee
Databricks Employee

‎09-08-2023 03:48 PM

@bzh 
Just a friendly follow-up. Did any of the responses help you to resolve your question? if it did, please mark it as best. Otherwise, please let us know if you still need help.

0 Kudos

nag_kanchan
New Contributor III

‎11-30-2023 05:04 AM

The solution did not work for me. It was throwing an error stating: raise Py4JError( py4j.protocol.Py4JError: An error occurred while calling o434.readStream. Trace: py4j.Py4JException: Method readStream([class java. util.ArrayList]) does not exist.

An alternate solution which worked for me was:

import dlt

@dlt.table 
def final_table(): 
   table1 = spark.table("table_1") 
   table2 = spark.table("table_2")
   return table1.union(table2)

@dlt.table
def target():
    return dlt.apply_changes(
        target="target_table",
        source="final_table",
        keys=["key_column"],
        sequence_by="abc",
        stored_as_scd_type="1"
    )

 

0 Kudos