Options
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
12-19-2024 02:26 AM
Final solution for the Bronze:
# Define view as the source
@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
)
# Define a DLT to store the bronze data with Change Data Feed enabled
dlt.create_streaming_table(
name="Transactions_Bronze", # No database qualifier in the table name
table_properties={
"quality": "bronze",
"delta.enableChangeDataFeed": "true"
}
)
# Implement SCD type 2
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
)