https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/ba-p/43937 <H3><STRONG>Introduction</STRONG></H3> <P><SPAN>In Part 1 of this blog series, we explored the various types of duplicates, considerations for remediation, and the impacts of unchecked duplicated records on strategic decision-making.&nbsp;</SPAN></P> <P><SPAN>If you want to learn more about this, please read: </SPAN><A href="https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/ba-p/40568" target="_blank" rel="noopener">How to implement Slowly Changing Dimensions when you have duplicates - Part 1: What to look out for?</A></P> <P><SPAN>In this blog, we will focus on leveraging </SPAN><A href="https://www.databricks.com/product/delta-live-tables" target="_blank" rel="noopener"><SPAN>Delta Live Tables</SPAN></A><SPAN> pipelines as a robust solution for handling duplicates and building an efficient data pipeline to maintain your Slowly Changing Dimensions.</SPAN></P> <H3><STRONG>Table of Contents</STRONG></H3> <UL> <LI><A href="#h_99442612331694082558192" target="_self">About your data pipeline</A></LI> <LI><A href="#h_253153268211694082566206" target="_self">Loading your Bronze table with Databricks Auto Loader</A></LI> <LI><A href="#h_85486507381694082572607" target="_self">Cleanse from other data quality issues in the Silver Layer</A></LI> <LI><A href="#h_964121769541694082585144" target="_self">Maintaining your slowly changing dimensions in your Gold Layer</A> <UL> <LI><A href="#h_277270263691694082591221" target="_self">Dropping Intra-batch duplicates</A></LI> <LI><A href="#h_420888401831694082598746" target="_self">Handling Inter-batch duplicates</A></LI> </UL> </LI> <LI><A href="#h_597995523961694082605592" target="_self">Configuring your Delta Live Tables pipeline</A></LI> <LI><A href="#h_7002050581081694082613529" target="_self">Let’s run some tests!</A> <UL> <LI><A href="#h_5934493061191694082620948" target="_self">Test #1 - Initial Load with Intra-Batch Exact Duplicates</A></LI> <LI><A href="#h_9513327451321694082627426" target="_self">Test #2 - Next batch with Inter-Batch, Exact Duplicates</A></LI> <LI><A href="#h_5875921001411694082634968" target="_self">Test #3 - Load with Intra- &amp; Inter- Batch Same Business Meaning Duplicates</A></LI> <LI><A href="#h_270263761491694082641326" target="_self">Test #4 - Load with Intra- &amp; Inter- Batch Out of Order Records</A></LI> </UL> </LI> <LI><A href="#h_4448571301561694082648876" target="_self">Conclusion</A><STRONG><BR /></STRONG></LI> </UL> <H3><STRONG>About your data pipeline</STRONG></H3> <P>Your data pipeline implements the <A href="https://www.databricks.com/glossary/medallion-architecture" target="_blank" rel="noopener">Medallion Architecture</A> which loads source data into a Bronze table, applies <A href="https://docs.databricks.com/delta-live-tables/expectations.html" target="_blank" rel="noopener">data quality rules and <I>expectations</I></A> in the Silver table, and maintains the slowly changing dimensions in a Gold table.</P> <P class="lia-align-center"><span class="lia-inline-image-display-wrapper lia-image-align-center" image-alt="Avnish_Jain_0-1694079325009.png" style="width: 720px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3547iA181B442688C5FDB/image-dimensions/720x236?v=v2" width="720" height="236" role="button" title="Avnish_Jain_0-1694079325009.png" alt="Avnish_Jain_0-1694079325009.png" /></span><FONT size="2"><I><SPAN>Figure 1 - High level solution architecture diagram of the sample data pipeline<BR /><BR /></SPAN></I></FONT></P> <H3><STRONG>Loading your Bronze table with Databricks Autoloader</STRONG></H3> <P><SPAN>The data pipeline begins with the incremental loading of source data with Databricks </SPAN><A href="https://docs.databricks.com/ingestion/auto-loader/index.html" target="_blank" rel="noopener"><SPAN>Auto Loader</SPAN></A><SPAN> into a Bronze table. The Bronze table acts as the initial landing zone for incoming </SPAN><A href="https://www.databricks.com/product/data-streaming" target="_blank" rel="noopener"><SPAN>streaming</SPAN></A><SPAN> data, where records are inserted into in an append-only fashion.&nbsp;</SPAN></P> <P><SPAN>To aid with the identification of all </SPAN><I><SPAN>kinds</SPAN></I><SPAN> of duplicates, we will append a new column called </SPAN><STRONG>“</STRONG><STRONG>data_hash</STRONG><STRONG>”</STRONG><SPAN> which we can specify to be calculated on all columns or a select subset. Records with the same </SPAN><SPAN>data_hash</SPAN><SPAN> value will indicate that they are duplicate records.&nbsp;</SPAN></P> <P><SPAN>If opting for a select subset of records - such as omitting technical columns to identify same business meaning duplicates - ensure you have determined and agreed this with your business stakeholders.</SPAN></P> <P>&nbsp;</P> <P>&nbsp;</P> <LI-CODE lang="python">import dlt from pyspark.sql.functions import concat_ws, md5, col, current_timestamp @dlt.table def bronze_customers(): return ( spark.readStream.format("cloudFiles") \ .option("cloudFiles.format", "csv") \ .option("cloudFiles.inferColumnTypes", "true") \ .option("header", "true") \ .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") \ .load("cloud-storage-source-data-directory") \ .withColumn("data_hash", \ md5(concat_ws('-', col("cust_id"), \ col("cust_name"), \ col("cust_location")))) \ .withColumn("file_name", col("_metadata.file_name")) \ .withColumn("insert_timestamp", current_timestamp()) )</LI-CODE> <P>&nbsp;</P> <P>&nbsp;</P> <P><SPAN>The Bronze table will store all data received from the source </SPAN><STRONG>including all duplicates received</STRONG><SPAN>. These duplicates need to be effectively processed and resolved in subsequent stages of the pipeline.<BR /><BR /></SPAN></P> <TABLE> <TBODY> <TR> <TD> <P><STRONG>Note!</STRONG> <SPAN><BR /></SPAN><SPAN><BR /></SPAN><SPAN>Ensure your chosen hash function is </SPAN><STRONG><I>deterministic</I></STRONG><SPAN> (i.e. given the same input, the same exact output value will always be generated).</SPAN><SPAN><BR /><BR /></SPAN></P> </TD> </TR> </TBODY> </TABLE> <H3><STRONG><BR />Cleanse from other data quality issues in the Silver Layer</STRONG></H3> <P><SPAN>The next step in the pipeline involves further data cleaning of records as </SPAN><I><SPAN>incrementally</SPAN></I><SPAN> received from the Bronze Table.&nbsp;</SPAN></P> <P><SPAN>Here, we apply business data quality (DQ) rules and perform data cleaning operations in the silver table with Delta Live Tables Expectations. Expectations provides you with a simple but powerful DQ framework that will enhance the overall reliability and trustworthiness of your data pipeline.</SPAN></P> <P>&nbsp;</P> <P>&nbsp;</P> <LI-CODE lang="python">@dlt.table(name="silver_customers") @dlt.expect_or_drop("valid_cust_id", "cust_id &gt; 0") def silver_customers(): return ( dlt.read_stream("bronze_customers") .select("*") )</LI-CODE> <P>&nbsp;</P> <P>&nbsp;</P> <P><SPAN>To read more about DLT Expectations - please refer to the following </SPAN><A href="https://docs.databricks.com/delta-live-tables/expectations.html" target="_blank" rel="noopener"><SPAN>documentation</SPAN></A><SPAN>. </SPAN></P> <H3><STRONG>Maintaining your slowly changing dimensions in your Gold Layer</STRONG></H3> <P><SPAN>The final stage of the data pipeline focuses on maintaining slowly changing dimensions in the Gold table which serves as the trusted source for historical analysis and decision-making. Here, we will remove the duplicates in 2 steps: first the intra-batch duplicates in a view, followed by the inter-batch duplicates.&nbsp;</SPAN></P> <H4><STRONG>Dropping Intra-batch duplicates</STRONG></H4> <P><SPAN>In the below code snippet, we are leveraging the native dropDuplicates() function and are passing in the previously calculated </SPAN><I><SPAN>data_hash </SPAN></I><SPAN>column to handle the intra-batch duplicates.</SPAN></P> <P><SPAN>This will drop any records with the same value in the</SPAN><I><SPAN> data_hash</SPAN></I><SPAN> column. As hashes are deterministic, we can safely remove intra-batch duplicate records including those we identified as having the same business meaning. However, we are still to handle inter-batch duplicates.&nbsp;</SPAN></P> <P>&nbsp;</P> <P>&nbsp;</P> <LI-CODE lang="python">@dlt.view def gold_customers_intra_dedup(): return ( dlt.read_stream("silver_customers") .select("*") .dropDuplicates(["data_hash"]) )</LI-CODE> <P>&nbsp;</P> <P>&nbsp;</P> <P><SPAN>As we have defined this as a DLT View, data will not be persisted.</SPAN></P> <H4><STRONG>Handling Inter-batch duplicates</STRONG></H4> <P><SPAN>To maintain your SCD Type 1 and Type 2 dimensions, Delta Live Tables provides the APPLY CHANGES INTO syntax which streamlines the process of synchronizing data and maintaining the integrity of the SCD tables. With minimal code, it can efficiently handle any INSERT, UPDATE and/or DELETE operations required when merging changes from the source table. This provides a seamless and scalable solution for managing evolving data in real-time.</SPAN></P> <P>&nbsp;</P> <P>&nbsp;</P> <LI-CODE lang="python">dlt.create_streaming_table("gold_customers") dlt.apply_changes( target = "gold_customers", source = "gold_customers_intra_dedup", keys = ["cust_id"], sequence_by = col("cust_last_updt_ts"), stored_as_scd_type = "2", track_history_column_list = ["data_hash"] )</LI-CODE> <P>&nbsp;</P> <P>&nbsp;</P> <P><SPAN>Let’s walk-through what’s happening in the above code-snippet:</SPAN></P> <UL> <LI><STRONG>target</STRONG><STRONG>:</STRONG><SPAN> the name of our Gold slowly changing dimension table.<BR /></SPAN></LI> <LI><STRONG>source</STRONG><STRONG>:</STRONG><SPAN> the name of our Silver cleansed table.</SPAN><SPAN><BR /></SPAN></LI> <LI><STRONG>keys</STRONG><SPAN>: the column(s) used for joining to identify matching (and unmatching) rows between the target and source tables. This will be the primary key(s) of the Gold/Silver table.<BR /></SPAN></LI> <LI><STRONG>sequence_by</STRONG><STRONG>: </STRONG><SPAN>the column that determines the logical order of records. DLT also uses this sequencing to handle change events that arrive out of order. Timestamps are commonly used such as the insert_timestamp (of when the record was ingested by your platform), or a business timestamp as received in the source data.</SPAN><SPAN><BR /></SPAN></LI> <LI><STRONG>stored_as_scd_type: </STRONG><SPAN>Either “1” or “2” to determine which SCD Type the target table will be maintained as.<BR /></SPAN></LI> <LI><SPAN><STRONG>track_history_column_list</STRONG><STRONG>:</STRONG> For SCD Type 2 tables, we can specify the column(s) that are used to track for history and trigger a changed record. This is the parameter that will help us handle any <I>inter-batch duplicates.&nbsp;</I><BR /></SPAN></LI> </UL> <P><SPAN>As discussed in </SPAN><A href="https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/ba-p/40568" target="_blank" rel="noopener"><SPAN>Part 1</SPAN></A><SPAN> of the blog series, there is no functional need to specifically handle inter-batch duplicates for SCD Type 1 tables as they do not maintain historical versions or track changes over time. However, inter-batch duplicates do need to be handled for SCD Type 2 tables in order to maintain data accuracy or integrity.&nbsp;</SPAN></P> <P><SPAN>To mitigate this, the </SPAN><STRONG>track_history_column_list</STRONG><SPAN> specified on the</SPAN><STRONG> data_hash</STRONG><SPAN> column tells DLT to only generate a changed,&nbsp;<EM>upserted</EM> record when there is a different value between source and target. If the values are the same, then just rather hard-update the record (similar to an SCD Type 1).<BR /><BR /></SPAN></P> <H3><STRONG>Configuring your Delta Live Tables pipeline</STRONG></H3> <P><SPAN>To use the </SPAN><STRONG>track_history_column_list</STRONG><SPAN> in Delta Live Tables, you must explicitly enable the feature in your pipeline by adding the following configuration to your Delta Live Tables pipeline settings.&nbsp;</SPAN></P> <P><SPAN>This can be done by adding in the below configuration in <STRONG>Advanced </STRONG>section of the Pipeline Settings in the UI.&nbsp;<BR /><BR /></SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-center" image-alt="Avnish_Jain_0-1694081511796.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3548i8F904F6E8B0EC812/image-size/large?v=v2&amp;px=999" role="button" title="Avnish_Jain_0-1694081511796.png" alt="Avnish_Jain_0-1694081511796.png" /></span></P> <P><SPAN><BR />You can also modify the DLT JSON configuration:<BR /><BR /></SPAN></P> <PRE><SPAN>{</SPAN><BR /><SPAN>&nbsp;&nbsp;</SPAN><SPAN>"configuration"</SPAN><SPAN>:</SPAN> <SPAN>{</SPAN><BR /><SPAN>&nbsp;&nbsp;&nbsp;&nbsp;</SPAN><SPAN>"pipelines.enableTrackHistory"</SPAN><SPAN>:</SPAN> <SPAN>"true"</SPAN><BR /><SPAN>&nbsp;&nbsp;</SPAN><SPAN>}</SPAN><BR /><SPAN>}</SPAN></PRE> <P><STRONG>And that is it!&nbsp;</STRONG></P> <P><SPAN>You can now save and run your DLT pipeline which will look a little like this!</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-center" image-alt="Avnish_Jain_0-1694081643022.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3549iC320A35A7ACDB043/image-size/large?v=v2&amp;px=999" role="button" title="Avnish_Jain_0-1694081643022.png" alt="Avnish_Jain_0-1694081643022.png" /></span></P> <H3><STRONG>Let’s run some tests!</STRONG></H3> <P><SPAN>In this section, we will analyze the outcomes of executing the Delta Live Tables (DLT) pipeline against each test case mentioned earlier, specifically focusing on the SCD Type 2 tables</SPAN><SPAN>.</SPAN></P> <H4><STRONG><BR />Test #1 - Initial Load with Intra-Batch Exact Duplicates</STRONG></H4> <P><SPAN>Our first test will perform the initial load of our Gold SCD Type 2 table with the first batch of data containing an <STRONG>exact, intra-batch duplicate for John Doe.<BR /><BR /></STRONG></SPAN></P> <PRE>cust_id,cust_name,cust_location,cust_last_updt_ts<BR />1,Jane Smith,United States,2023-03-02 12:31:53<BR />2,John Doe,Australia,2023-03-02 12:31:53<BR />2,John Doe,Australia,2023-03-02 12:31:53</PRE> <P><SPAN>After successful execution of the above code, we can see that DLT has only loaded the correct two records.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-left" image-alt="Avnish_Jain_0-1694081886071.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3550i1B34F50095A19208/image-size/large?v=v2&amp;px=999" role="button" title="Avnish_Jain_0-1694081886071.png" alt="Avnish_Jain_0-1694081886071.png" /></span></P> <H4>&nbsp;</H4> <H4>&nbsp;</H4> <H4>&nbsp;</H4> <H4><STRONG><BR />Test #2 - Next batch with Inter-Batch, Exact Duplicates</STRONG></H4> <P><SPAN>Our second batch builds on after the successful completion of the first test but this time we will be loading the below batch of data containing an<STRONG> inter-batch, exact duplicate for Jane Smith;</STRONG> and a new record for Angela.<BR /><BR /></SPAN></P> <PRE><FONT size="2">cust_id,cust_name,cust_location,cust_last_updt_ts</FONT><BR /><FONT size="2">1,Jane Smith,United States,2023-03-02 12:31:53</FONT><BR /><FONT size="2">3,Angela Hamilton,England,2023-03-02 12:31:59</FONT></PRE> <P><SPAN>After successful execution of the above code, we can see that DLT has appropriately handled the duplicate record for Jane.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="Avnish_Jain_1-1694082079933.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3551iBE3F99AF295F4924/image-size/large?v=v2&amp;px=999" role="button" title="Avnish_Jain_1-1694082079933.png" alt="Avnish_Jain_1-1694082079933.png" /></span></P> <H4><STRONG><BR />Test #3 - Load with Intra- &amp; Inter- Batch Same Business Meaning Duplicates</STRONG></H4> <P><SPAN>Our next batch continues this time with </SPAN><STRONG>intra-</STRONG><SPAN> and</SPAN><STRONG> inter-batch duplicates</STRONG><SPAN> where we have received records that are not exact duplicates <STRONG>as the </STRONG></SPAN><FONT face="courier new,courier">cust_last_updt_ts</FONT><STRONG> column holds different values.<BR /><BR /></STRONG></P> <PRE>cust_id,cust_name,cust_location,cust_last_updt_ts<BR />3,Angela Hamilton,England,2023-03-02 13:32:00<BR />4,Sachin Singh,India,2023-03-02 13:31:59<BR />4,Sachin Singh,India,2023-03-02 13:32:00</PRE> <P><SPAN>After agreeing with the business that a change only to technical columns, such as </SPAN><FONT face="courier new,courier"><SPAN>cust_last_updt_ts</SPAN></FONT><SPAN>, our DLT pipeline does not trigger a new ‘change’ record in our SCD Type 2 Gold Table for Angela and only ingests the additional single record for Sachin.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="Avnish_Jain_0-1694082297092.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3552i8D894FDD1285DAA7/image-size/large?v=v2&amp;px=999" role="button" title="Avnish_Jain_0-1694082297092.png" alt="Avnish_Jain_0-1694082297092.png" /></span></P> <H4><STRONG><BR />Test #4 - Load with Intra- &amp; Inter- Batch Out of Order Records</STRONG></H4> <P><SPAN>Our final batch does not consist of duplicates but rather </SPAN><STRONG>intra-</STRONG><SPAN> and</SPAN><STRONG> inter-batch </STRONG><SPAN>records that have been <STRONG>received out of order.</STRONG> Not managing these edge-cases correctly could lead to severe issues in data quality that could significantly impact decision making. </SPAN></P> <P><SPAN>In the example below, we receive a </SPAN><I><SPAN>previous</SPAN></I><SPAN> record for Jane Smith location prior to moving to the United States and two records for Sachin - a move to Indonesia </SPAN><I><SPAN>before</SPAN></I><SPAN> his move to India, and a move after to Australia.<BR /><BR /></SPAN></P> <PRE><SPAN>cust_id,cust_name,cust_location,cust_last_updt_ts<BR />1,Jane Smith,England,2022-01-01 00:00:00<BR />4,Sachin Singh,Indonesia,2022-01-01 00:00:00<BR />4,Sachin Singh,Australia,2023-06-01 00:00:00</SPAN></PRE> <P><SPAN>As you can see from the result set below, DLT was able to remediate and stitch these records correctly into the Gold Table!</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="Avnish_Jain_1-1694082386133.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3553i21AB91A571004E77/image-size/large?v=v2&amp;px=999" role="button" title="Avnish_Jain_1-1694082386133.png" alt="Avnish_Jain_1-1694082386133.png" /></span></P> <H3><STRONG><BR />Conclusion</STRONG></H3> <P><SPAN>In conclusion, leveraging </SPAN><A href="https://www.databricks.com/product/delta-live-tables" target="_blank" rel="noopener"><SPAN>Delta Live Tables</SPAN></A><SPAN> enhances the robustness of your data pipelines and simplifies the management of duplicate records when implementing your slowly changing dimensions. With its provided APPLY CHANGES INTO syntax, you can effectively handle intra-batch, inter-batch, and same business meaning duplicates as well as out-of-order records that could further hinder data quality. With clean and accurate dimensions, organizations can unlock the full potential of their data to enable accurate historical analysis, data-driven decision-making, and better foster trust in the data.</SPAN></P> <H3 id="toc-hId-1418975055"><STRONG>Coming up next!</STRONG><SPAN><BR /></SPAN></H3> <P><SPAN>In Part 3 of this blog series, we will walk through a practical implementation of a <A href="https://www.databricks.com/product/data-streaming" target="_self">Spark Structured Streaming</A> data pipeline exposed to the same challenges!</SPAN></P> Fri, 29 Sep 2023 19:39:47 GMT Avnish_Jain 2023-09-29T19:39:47Z https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/ba-p/43937 <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="pexels-edward-jenner-4252526.jpg" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/3557iF2B055CDFAE7A55C/image-size/large?v=v2&amp;px=999" role="button" title="pexels-edward-jenner-4252526.jpg" alt="pexels-edward-jenner-4252526.jpg" /></span></P> <P>&nbsp;</P> Fri, 29 Sep 2023 19:39:47 GMT https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/ba-p/43937 Avnish_Jain 2023-09-29T19:39:47Z https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/bc-p/50139#M64 <P>Hey&nbsp;<a href="https://community.databricks.com/t5/user/viewprofilepage/user-id/46052">@Avnish_Jain</a>,&nbsp;we're implementing a very similar inter and intra-batch deduplication process albeit with SCD type-1. However,&nbsp; we are afraid the drop_duplicates() (in your case dropDuplicates(["data_hash"])) might be looking at the whole stream of data as it could be stateful. How do we ensure, the drop_duplicates is only looking at the data inside the microbatch?</P> Tue, 31 Oct 2023 02:17:00 GMT https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/bc-p/50139#M64 sandeep_gunda 2023-10-31T02:17:00Z https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/bc-p/50924#M70 <P>Thank you for this, when is Part 3 which includes Spark Structured Streaming due?&nbsp;</P> Sat, 11 Nov 2023 05:46:28 GMT https://community.databricks.com/t5/technical-blog/how-to-implement-slowly-changing-dimensions-when-you-have/bc-p/50924#M70 rantav 2023-11-11T05:46:28Z