import dlt
from pyspark.sql.functions import col, lit, current_timestamp
from pyspark.sql import functions as F

@dlt.table(
    name="Transactions_Bronze_Raw",
    table_properties={
        "delta.enableChangeDataFeed": "true",
        "quality": "bronze"
    }
)

def Transactions_Bronze_Raw():
    # Read streaming data from the landing path (source of data)
    df = (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("inferSchema", True)
        .option("cloudFiles.inferColumnTypes", "true")
        .option("recursiveFileLookup", "true")
        .load(landing_json_path)
    )

    # Add metadata for inserted time
    df = df.withColumn("SDA_Inserted", F.date_format(F.current_timestamp(), "yyyy-MM-dd HH:mm:ss"))
    return df

# Transactions_Bronze
dlt.create_streaming_table(
    name="Transactions_Bronze",  # No database qualifier in the table name
    table_properties={
        "quality": "silver",
        "delta.enableChangeDataFeed": "true"
    }
)

# Apply change data capture logic to handle SCD Type 2 for the streaming table
dlt.apply_changes(
    target="Transactions_Bronze",  
    source="Transactions_Bronze_Raw",  
    keys=["Company", "VoucherType", "AccountingYear", "VoucherNo", "RowNo"],
    sequence_by=F.col("SDA_Inserted"),  
    stored_as_scd_type=2  
)

Silver layer:

@dlt.table(
    name="Transactions_Silver",
    table_properties={
        "quality": "silver",
        "delta.enableChangeDataFeed": "true"
    }
)

def Transactions_Silver():
    # Read from the Bronze table
    bronze_df = spark.readStream.format("delta").table("sda_edw_bronze.erp.Transactions_Bronze")

    # Filter only open rows
    silver_df = bronze_df.filter(F.col("__END_AT").isNull())

    # Add a new SDA_Inserted column
    silver_df = silver_df.withColumn(
        "SDA_Inserted", 
        F.date_format(F.current_timestamp(), "yyyy-MM-dd HH:mm:ss")
    )

    # Return the transformed DataFrame
    return silver_df

 The issue is that the bronze layer fails, 
"message": "Update ff5d14 is FAILED.",
"level": "ERROR",
"error": {
"exceptions": [
{
"message": "INTERNAL_ERROR: Pipeline cluster is not healthy."
}
]

But the cluster is fine, because I can run other notebooks without issues.

Do you see any issue with my code for the bronze layer?

First I wanted to create a view (as you suggested @filipniziol for the previous issue) of the input data and use that as a source for the bronze table, but that didnt work either. The pipeline ran without any issue, but the data dint get ingested. This is how that ide looked:

@dlt.view
def Transactions_Bronze_View():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("inferSchema", True)
        .option("cloudFiles.inferColumnTypes", "true")
        .option("recursiveFileLookup", "true")
        .load(landing_json_path)
        .withColumn("SDA_Inserted", F.date_format(F.current_timestamp(), "yyyy-MM-dd HH:mm:ss"))  # Add metadata
    )

# Transactions_Bronze
dlt.create_streaming_table(
    name="Transactions_Bronze",  # No database qualifier in the table name
    table_properties={
        "quality": "bronze",
        "delta.enableChangeDataFeed": "true"
    }
)

# Use the view as the source for SCD2 table creation
dlt.apply_changes(
    target="Transactions_Bronze",
    source="Transactions_Bronze_View",
    keys=["Company", "VoucherType", "AccountingYear", "VoucherNo", "RowNo"],
    sequence_by=F.col("SDA_Inserted"),  # Timestamp column to sequence updates
    stored_as_scd_type=2
)