Dharinip
Contributor

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?

 

 

DHARINI PARGUNAN