topic Need help with DLT Pipeline in Data Engineering

Need help with DLT Pipeline

Fatimah-Tariq — Wed, 23 Oct 2024 07:24:34 GMT

I have a DLT pipeline running daily for months and recently found out one issue in my silver layer code and as a result of that, now I have faulty data in my silver schema. Please note that the tables in Silver schema are streaming tables handled within the context of DLT.
I want to delete the faulty records from my silver schema and keep the pipeline running with the correct code as it's running normally. Could someone please suggest me what would be the best possible approach for me to update my silver schema streaming tables by deleting faulty records? Early response would be highly appreciated.

Re: Need help with DLT Pipeline

saurabh18cs — Wed, 23 Oct 2024 08:14:31 GMT

Use Delta Lake's DELETE command to remove the faulty records from your Silver tables. You can do this in a Databricks notebook or a separate script.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("DeleteFaultyRecords").getOrCreate()

# Define the criteria for faulty records
faulty_criteria = "your_faulty_criteria_here"

# List of Silver tables to clean
silver_tables = ["silver_table1", "silver_table2"]

for table in silver_tables:
spark.sql(f"DELETE FROM {table} WHERE {faulty_criteria}")

Also, Ensure that your DLT pipeline code is corrected to prevent future faulty records. Update the transformation logic in your Silver layer to handle the data correctly. (spark dataframe transformations or constraints)

Re: Need help with DLT Pipeline

AngadSingh — Wed, 23 Oct 2024 08:26:43 GMT

Hi Fatimah,

You can delete the records from silver layer as long as those records don't get reloaded again (from bronze). More info is here

May I also understand that are you are using CDC apply_changes method for loading data to silver (like SCD 1)? If not, definitely above can be done.

Cheers

Re: Need help with DLT Pipeline

Fatimah-Tariq — Wed, 23 Oct 2024 08:33:05 GMT

Yes, I'm using CDC apply_changes method with SCD type 1

Re: Need help with DLT Pipeline

AngadSingh — Fri, 25 Oct 2024 21:04:08 GMT

In that case, what's the expression in the "apply_as_delete" option? Or please share your CDC apply_changes code block?

Re: Need help with DLT Pipeline

Fatimah-Tariq — Mon, 28 Oct 2024 06:37:29 GMT

I do not have "apply_as_delete" option.
Here's my apply_changes code

dlt.apply_changes(

target = silver_table,

source = bronze_dlt_view,

keys = primary_keys,

sequence_by = col(sequence_col),

stored_as_scd_type = 1,

except_column_list = ["extract_datetime_utc", "_rescued_data", "dlt_extract_datetime_utc"]

)