- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
03-24-2025 01:42 PM
My source table is a streaming table and I did not perform complex transformation.
1. Reading the source table
2. Filtering the data (only latest records)
3. Performing column name changes.
My target table is Materialized view:
def my_incremental_view():
# STEP 1 - To read only the active records
src_silver_df = spark.read.table(f"{catalog}.{src_schema}.{src_table}")
src_silver_df = src_silver_df.filter(src_silver_df.__END_AT.isNull())
# STEP 2 - To read only the latest records out of activ erecords
df = spark.sql("SELECT StartDate FROM c_realestate_dev.gold.goldtables WHERE TableName = 'modeling_incre'")
last_timestamp = df.first()['StartDate']
latest_records_source_df = src_silver_df.filter(src_silver_df.__START_AT >= last_timestamp)
# STEP 3 - To transform only the latest records
transformed_df = table_ColumnName_Initial_transformation(latest_records_source_df)
return transformed_df
@Dlt.table( name= final_table,
schema = schema_enrichment,
comment=f"This table contains information about all the Enrichment Location summary Information for each Client",
spark_conf = { "pipelines.incompatibleViewCheck.enabled" : "false" },
table_properties={
"quality": "gold",
"pipelines.autoOptimize.managed": "true"
})
def load_silver_data():
base_table = spark.read.table("my_incremental_view")
return base_table
How to perform the incremental load?