Issue 180 : provide apply_changes_from_snapshot support in silver layer

[ravi-databricks]

Introduced new attribute silver_apply_changes_from_snapshot for silver snapshot api in onboarding file.
e.g.

 { "data_flow_id": "201", "data_flow_group": "A1", "source_system": "delta", "source_format": "snapshot", "source_details": { "snapshot_format": "delta", "source_catalog_it": "{uc_catalog_name}", "source_table": "source_products_delta", "source_database":"{source_database}" }, "bronze_database_it": "{uc_catalog_name}.{bronze_schema}", "bronze_table": "products", "bronze_apply_changes_from_snapshot": { "keys": [ "product_id" ], "scd_type": "2" }, "silver_catalog_it": "{uc_catalog_name}", "silver_database_it": "{silver_schema}", "silver_table": "products", "silver_table_comment": "products silver table", "silver_apply_changes_from_snapshot":{ "keys": [ "product_id" ], "scd_type": "2" }, "silver_transformation_json_it": "{uc_volume_path}/integration_tests/conf/silver_transformations_snapshot.json" } 

For silver layer bronze table is source which is delta so no need to provide custom snapshot function, as API will point to bronze table as source.
In below example source will be {bronze_database_it}.{bronze_table}

 dlt.apply_changes_from_snapshot( target=target_table, source=source, keys=self.appy_changes_from_snapshot.keys, stored_as_scd_type=self.appy_changes_from_snapshot.scd_type, track_history_column_list=self.appy_changes_from_snapshot.track_history_column_list, track_history_except_column_list=self.appy_changes_from_snapshot.track_history_except_column_list, ) 

If you want to add your own customization then provide function inside init_dlt_meta_pipeline.py notebook as below

DataflowPipeline.invoke_dlt_pipeline(spark, layer, bronze_next_snapshot_and_version=next_snapshot_and_version_bronze_function, silver_next_snapshot_and_version=next_snapshot_and_version_silver_function)

Please check demo here

[brij-raghuwanshi-db]

Approved.