https://community.databricks.com/t5/technical-blog/deep-dive-streaming-deduplication/ba-p/105062 <P><SPAN>In this article we will cover in depth about </SPAN><A href="https://spark.apache.org/docs/3.5.3/structured-streaming-programming-guide.html#streaming-deduplication" target="_blank" rel="noopener"><SPAN>streaming deduplication</SPAN></A><SPAN> using </SPAN><A href="https://spark.apache.org/docs/3.5.3/structured-streaming-programming-guide.html#handling-late-data-and-watermarking" target="_blank" rel="noopener"><SPAN>watermarking</SPAN></A><SPAN> with </SPAN><A href="https://jaceklaskowski.gitbooks.io/spark-structured-streaming/content/spark-sql-streaming-Dataset-dropDuplicates.html" target="_blank" rel="noopener"><SPAN>dropDuplicates</SPAN></A><SPAN> and </SPAN><A href="https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.dropDuplicatesWithinWatermark.html" target="_blank" rel="noopener"><SPAN>dropDuplicatesWithinWatermark</SPAN></A><SPAN>, how they are different. This blog expects you to have a good understanding on how </SPAN><A href="https://www.databricks.com/blog/feature-deep-dive-watermarking-apache-spark-structured-streaming" target="_blank" rel="noopener"><SPAN>watermarking</SPAN></A><SPAN> works in Spark Structured streaming. Lets dive right in.</SPAN></P> <P>&nbsp;</P> <H1><SPAN>dropDuplicates</SPAN></H1> <P><SPAN>When </SPAN><I><SPAN>dropDuplicates</SPAN></I><SPAN> is used in conjunction with </SPAN><I><SPAN>withWatermark</SPAN></I><SPAN> operator, it maintains a global state, the stream is unbounded and will keep all the keys across triggers, i.e., global deduplication. Whatever watermark is specified using </SPAN><I><SPAN>withWatermark</SPAN></I><SPAN> makes the stream to filter out the late arriving records, i.e., records with </SPAN><I><SPAN>event_time</SPAN></I> <I><SPAN>T</SPAN></I><SPAN> get discarded at the beginning of the micro-batch if the watermark is greater than or equal to </SPAN><I><SPAN>T</SPAN></I><SPAN>. This approach is generally not recommended; an unbounded state means that the state store grows continuously, potentially causing out-of-memory issues. Let's look at an example.&nbsp;</SPAN></P> <P><SPAN>Create a source table to stream from:</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">drop table if exists mt_asqs.demo.dd_source; create table mt_asqs.demo.dd_source ( event_id string, event_time timestamp )</LI-CODE> <P>&nbsp;</P> <P><SPAN>Streaming code:&nbsp;</SPAN><SPAN>Note that we are using </SPAN><I><SPAN>event_time</SPAN></I><SPAN> as the watermarking column and not including the </SPAN><I><SPAN>event_time</SPAN></I><SPAN> column as deduplication key in </SPAN><I><SPAN>dropDuplicates</SPAN></I><SPAN> function.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">volume_path = '/Volumes/mt_asqs/demo/checkpoint_paths' stream_df = ( spark.readStream.format("delta") .table("mt_asqs.demo.dd_source") .withWatermark("event_time", "5 minutes") .dropDuplicates(["event_id"]) ) ( stream_df.writeStream .format("delta") .option("checkpointLocation", f"{volume_path}/dd_test3") .table("mt_asqs.demo.dd_output") )</LI-CODE> <P>&nbsp;</P> <P><SPAN>Inserting records into streaming source table&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source values ('A', to_timestamp('2024-01-01 10:00:00')); insert into dd_source values ('B', to_timestamp('2024-01-01 10:00:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Since the watermark delay is set to 5 minutes, above records set the watermark to </SPAN><I><SPAN>9:55.&nbsp; </SPAN></I><SPAN>The state store would have 2 keys A &amp; B.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="markup">insert into dd_source values ('C', to_timestamp('2024-01-01 10:07:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Watermarking advances to 10:02, C gets inserted into the state store. Ideally since the watermarking advanced to 10:02, Keys A &amp; B should get evicted from the state store but that doesn’t happen with </SPAN><I><SPAN>dropDuplicates</SPAN></I><SPAN>, this makes global deduplication possible. Viewing the contents of the state store.</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">display(spark.read.format("statestore").load("/Volumes/mt_asqs/demo/checkpoint_paths/dd_test3"))</LI-CODE> <P>&nbsp;</P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_0-1736447301690.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14014i7983ED4927C0D445/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_0-1736447301690.png" alt="MuraliTalluri_0-1736447301690.png" /></span><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_1-1736447301693.png" style="width: 303px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14012i7989C263907C8405/image-dimensions/303x168?v=v2" width="303" height="168" role="button" title="MuraliTalluri_1-1736447301693.png" alt="MuraliTalluri_1-1736447301693.png" /></span></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source values ('A', to_timestamp('2024-01-01 10:01:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Since watermarking is set to 10:02, the above record A gets discarded at the beginning of the micro-batch. At this point record A is not even compared to state store for deduplication since it's discarded prior to that.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source values ('D', to_timestamp('2024-01-01 10:10:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Advances the watermark to 10:05, D gets appended to the state store.&nbsp;&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source values ('A', to_timestamp('2024-01-01 10:12:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>The above record gets checked against the watermark, since it's not a late arriving record it won’t be discarded. This record advances watermarking time to 10:07, but since the key(A) already exists in the state store it gets discarded in the deduplication process.</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source values ('E', to_timestamp('2024-01-01 10:07:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>E gets discarded due to the watermark(10:07). Contents of the streaming output table shows that there are no duplicate </SPAN><I><SPAN>event_ids</SPAN></I><SPAN>.&nbsp;</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_2-1736447301693.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14013i2CA2C80C2877983B/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_2-1736447301693.png" alt="MuraliTalluri_2-1736447301693.png" /></span></P> <H1><SPAN>dropDuplicatesWithinWatermark</SPAN></H1> <P><I><SPAN>dropDuplicatesWithinWatermark</SPAN></I><SPAN> does exactly what you would expect it to do, this is what you should use for deduplication when you know the time window within which you expect duplicate records in your stream. When </SPAN><I><SPAN>dropDuplicatesWithinWatermark</SPAN></I><SPAN> is used in conjunction with </SPAN><I><SPAN>withWatermark </SPAN></I><SPAN>delay </SPAN><I><SPAN>D</SPAN></I><SPAN>, a record with event_time </SPAN><I><SPAN>T</SPAN></I><SPAN> gets removed from the state store only when the watermark is greater than or equal to </SPAN><I><SPAN>T</SPAN></I><SPAN> + </SPAN><I><SPAN>D</SPAN></I><SPAN>. But the record will be discarded(late arriving) at the beginning of the micro-batch if the watermark is greater than or equal to </SPAN><I><SPAN>T</SPAN></I><SPAN>. Let's look at an example.&nbsp;</SPAN></P> <P><SPAN>Create a source table to stream from:</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">drop table if exists mt_asqs.demo.ddww_source; create table mt_asqs.demo.ddww_source ( event_id string, event_time timestamp )</LI-CODE> <P>&nbsp;</P> <P><SPAN>Streaming code:&nbsp;</SPAN><SPAN>It’s basically the same as above but using the </SPAN><I><SPAN>dropDuplicatesWithinWatermark</SPAN></I><SPAN> instead of </SPAN><I><SPAN>dropDuplicates</SPAN></I><SPAN>.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">stream_df = ( spark.readStream.format("delta") .table("mt_asqs.demo.ddww_source") .withWatermark("event_time", "5 minutes") .dropDuplicatesWithinWatermark(["event_id"]) ) volume_path = '/Volumes/mt_asqs/demo/checkpoint_paths' ( stream_df.writeStream .format("delta") .option("checkpointLocation", f"{volume_path}/ddww_test") .table("mt_asqs.demo.ddww_output") )</LI-CODE> <P>&nbsp;</P> <P><SPAN>Inserting records into streaming source table&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into ddww_source values('A',to_timestamp('2024-01-01 10:00:00')); insert into ddww_source values('B',to_timestamp('2024-01-01 10:00:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Since the watermarking delay is set to 5 minutes, the watermark advances to 9:55. The state store would have 2 keys A &amp; B.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into ddww_source values('C',to_timestamp('2024-01-01 10:07:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Watermark advances to 10:02, C gets inserted into the state store. At the end of every micro-batch, existing records in the state store are checked if the watermark is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN> + watermark delay (5 minutes). So both A &amp; B prevail.</SPAN></P> <P><SPAN>Viewing the contents of the state store, you can see that every key now has expiration time.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">display(spark.read.format("statestore").load("/Volumes/mt_asqs/demo/checkpoint_paths/ddww_test"))</LI-CODE> <P>&nbsp;</P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_3-1736447301693.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14015iCFDF4454BBFAB10D/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_3-1736447301693.png" alt="MuraliTalluri_3-1736447301693.png" /></span><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_4-1736447301693.png" style="width: 328px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14017iA8434F295107DD23/image-dimensions/328x188?v=v2" width="328" height="188" role="button" title="MuraliTalluri_4-1736447301693.png" alt="MuraliTalluri_4-1736447301693.png" /></span></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into ddww_source values('A',to_timestamp('2024-01-01 10:01:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Since watermarking is set to 10:02, the above record A gets discarded at the beginning of the micro-batch. At this point record A is not even compared to the state store for deduplication since it's discarded prior to that.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into ddww_source values('D',to_timestamp('2024-01-01 10:09:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Watermark advances to 10:04, D gets inserted into the state store. Existing records in the state store are checked if the watermark is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN> + watermark delay (5 minutes). So all the records prevail.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_5-1736447301691.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14016i09B05DC263C1D751/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_5-1736447301691.png" alt="MuraliTalluri_5-1736447301691.png" /></span><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_6-1736447301694.png" style="width: 359px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14018i15791C60446AC55E/image-dimensions/359x216?v=v2" width="359" height="216" role="button" title="MuraliTalluri_6-1736447301694.png" alt="MuraliTalluri_6-1736447301694.png" /></span></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into ddww_source values('A',to_timestamp('2024-01-01 10:12:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Watermark advances to 10:07. Since A already exists in the state store, the new record is considered as duplicate and gets discarded. Existing records in the state store are checked if the watermark(10:07) is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN> + watermark delay (5 minutes). So C &amp; D prevails, A &amp; B gets evicted from the state store.&nbsp;</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_7-1736447301693.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14020i36C8A3E8FF2D1117/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_7-1736447301693.png" alt="MuraliTalluri_7-1736447301693.png" /></span><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_8-1736447301692.png" style="width: 331px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14019i98330AA333F3774B/image-dimensions/331x196?v=v2" width="331" height="196" role="button" title="MuraliTalluri_8-1736447301692.png" alt="MuraliTalluri_8-1736447301692.png" /></span></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into ddww_source values('E',to_timestamp('2024-01-01 10:17:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Watermark advances to 10:12, E gets inserted into the state store. Existing records in the state store are checked if the watermark is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN> + watermark delay (5 minutes). So D prevails and C gets evicted from the state store.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_9-1736447301690.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14021iFA120A71C41A5738/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_9-1736447301690.png" alt="MuraliTalluri_9-1736447301690.png" /></span><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_10-1736447301692.png" style="width: 352px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14022iBFBAC43F5712AAE1/image-dimensions/352x183?v=v2" width="352" height="183" role="button" title="MuraliTalluri_10-1736447301692.png" alt="MuraliTalluri_10-1736447301692.png" /></span></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into ddww_source values('C',to_timestamp('2024-01-01 10:13:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>The watermark still remains at 10:12. Since C doesn’t exist in the state store, it won’t be considered a duplicate record. So C gets inserted into the state store. Existing records in the state store are checked if the watermark is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN> + watermark delay (5 minutes). So D &amp; E prevails.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_11-1736447301691.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14023i54861B73830A0EF9/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_11-1736447301691.png" alt="MuraliTalluri_11-1736447301691.png" /></span><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_12-1736447301691.png" style="width: 332px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14024i96BF31732D61D45B/image-dimensions/332x170?v=v2" width="332" height="170" role="button" title="MuraliTalluri_12-1736447301691.png" alt="MuraliTalluri_12-1736447301691.png" /></span></P> <P><SPAN>Contents of the streaming output table shows that there is one duplicate entry for </SPAN><I><SPAN>event_id</SPAN></I><SPAN> C. This would have been avoided if you had a larger watermark window, like 10 minutes.&nbsp;</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_13-1736447301691.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14026iFF48D36A7687F792/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_13-1736447301691.png" alt="MuraliTalluri_13-1736447301691.png" /></span></P> <H1><SPAN>dropDuplicates - </SPAN><I><SPAN>event_time </SPAN></I><SPAN>as deduplication key</SPAN></H1> <P><SPAN>We covered above where you specify </SPAN><I><SPAN>event_time </SPAN></I><SPAN>as the watermarking column but don’t include it as a deduplication key(parameter) in the </SPAN><I><SPAN>dropDuplicates. </SPAN></I><SPAN>But when event_time is specified as one of the deduplication keys, </SPAN><I><SPAN>dropDuplicates </SPAN></I><SPAN>won’t maintain an unbounded state, i.e., keys get evicted from the state store. If you set </SPAN><I><SPAN>withWatermark</SPAN></I><SPAN> delay </SPAN><I><SPAN>D,</SPAN></I><SPAN> a record with event_time </SPAN><I><SPAN>T</SPAN></I><SPAN> gets removed from the state store when the watermark is greater than or equal to </SPAN><I><SPAN>T</SPAN></I><SPAN>. Records will be checked at the beginning of the micro-batch execution and discarded(late arriving) if the watermark is greater than or equal to </SPAN><I><SPAN>T</SPAN></I><SPAN>. Let's look at an example.&nbsp;</SPAN></P> <P><SPAN>Create a source table to stream from:</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">drop table if exists mt_asqs.demo.dd_source2; create table mt_asqs.demo.dd_source2 ( event_id string, event_time timestamp )</LI-CODE> <P>&nbsp;</P> <P><SPAN>Streaming code:&nbsp;</SPAN><SPAN>Note that we are using </SPAN><I><SPAN>event_time</SPAN></I><SPAN> as the watermarking column and it's also used as the deduplication key in the dropDuplicates function.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">stream_df = ( spark.readStream.format("delta") .table("mt_asqs.demo.dd_source2") .withWatermark("event_time", "5 minutes") .dropDuplicates(["event_id", "event_time"]) ) volume_path = '/Volumes/mt_asqs/demo/checkpoint_paths' ( stream_df.writeStream .format("delta") .option("checkpointLocation", f"{volume_path}/dd_test2") .table("mt_asqs.demo.dd_output2") )</LI-CODE> <P>&nbsp;</P> <P><SPAN>Inserting records into streaming source table&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source2 values ('A',to_timestamp('2024-01-01 10:00:00')); insert into dd_source2 values ('B',to_timestamp('2024-01-01 10:00:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Since the watermarking delay is set to 5 minutes, above records set the watermark to </SPAN><I><SPAN>9:55.&nbsp; </SPAN></I><SPAN>The state store would have 2 records, A &amp; B.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source2 values ('C',to_timestamp('2024-01-01 10:07:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Watermark advances to 10:02, C gets inserted into the state store. At the end of every micro-batch, existing records in the state store are checked if the watermark is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN>. So both A &amp; B get evicted from the state store.</SPAN></P> <P><SPAN>Viewing the contents of the state store.</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">display(spark.read.format("statestore").load("/Volumes/mt_asqs/demo/checkpoint_paths/dd_test2"))</LI-CODE> <P>&nbsp;</P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_14-1736447301693.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14025i606D42C56C7DD1D8/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_14-1736447301693.png" alt="MuraliTalluri_14-1736447301693.png" /></span><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_15-1736447301691.png" style="width: 334px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14027i51EB48FAC18A2F97/image-dimensions/334x203?v=v2" width="334" height="203" role="button" title="MuraliTalluri_15-1736447301691.png" alt="MuraliTalluri_15-1736447301691.png" /></span></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source2 values ('A',to_timestamp('2024-01-01 10:01:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Since watermarking is set to 10:02, the above record A gets discarded at the beginning of the micro-batch. At this point record A is not even compared to state store for deduplication since it's discarded prior to that.&nbsp;</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source2 values ('C',to_timestamp('2024-01-01 10:09:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Advances the watermark to 10:04. C gets appended to the state store since </SPAN><I><SPAN>event_time</SPAN></I><SPAN> is also a deduplication key. The state store will have multiple rows for C.&nbsp;&nbsp;</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_16-1736447301692.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14028i3FFCD2F5A7EBD547/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_16-1736447301692.png" alt="MuraliTalluri_16-1736447301692.png" /></span></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source2 values ('D',to_timestamp('2024-01-01 10:10:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Advances the watermark to 10:05, D gets appended to the state store. Existing records in the state store are checked if the watermark is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN>. Both the entries for C prevails.&nbsp; The state store will have 3 entries(2 for C and 1 for D).</SPAN></P> <P>&nbsp;</P> <LI-CODE lang="python">insert into dd_source2 values ('A',to_timestamp('2024-01-01 10:12:00'));</LI-CODE> <P>&nbsp;</P> <P><SPAN>Watermark advances to 10:07, A gets inserted into the state store. Existing records in the state store are checked if the watermark is greater than or equal to </SPAN><I><SPAN>event_time</SPAN></I><SPAN>. One of the entries for C (event_time 10:07) gets evicted while the other prevails.</SPAN></P> <P><SPAN>Viewing the contents of the state store.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_17-1736447301692.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14029i0B93E2B978457D52/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_17-1736447301692.png" alt="MuraliTalluri_17-1736447301692.png" /></span></P> <P><SPAN>Contents of the streaming output table.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="MuraliTalluri_18-1736447301693.png" style="width: 400px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14030iF43F2BACC0CC1E95/image-size/medium?v=v2&amp;px=400" role="button" title="MuraliTalluri_18-1736447301693.png" alt="MuraliTalluri_18-1736447301693.png" /></span></P> <H1><SPAN>Conclusion</SPAN></H1> <P><SPAN>Each of these approaches solve different problems.&nbsp;</SPAN></P> <TABLE> <TBODY> <TR> <TD> <P><SPAN>Approach</SPAN></P> </TD> <TD> <P><SPAN>Scenario</SPAN></P> </TD> <TD> <P><SPAN>State store behavior</SPAN></P> </TD> </TR> <TR> <TD> <P><I><SPAN>dropDuplicates</SPAN></I><SPAN> + </SPAN><I><SPAN>withWatermark</SPAN></I></P> </TD> <TD> <P><SPAN>Global deduplication and don’t expect the duplicates with the same </SPAN><I><SPAN>event_time</SPAN></I></P> </TD> <TD> <P><SPAN>- Keep all the keys in the state store.</SPAN></P> <P><SPAN>- The state store grows continuously, potentially causing out-of-memory issues.</SPAN></P> </TD> </TR> <TR> <TD> <P><I><SPAN>dropDuplicatesWithinWatermark </SPAN></I><SPAN>+ </SPAN><I><SPAN>withWatermark</SPAN></I></P> </TD> <TD> <P><SPAN>- When you know the interval of time within which you might have duplicate records.</SPAN></P> <P><SPAN>- Deduplication within the specified time interval.&nbsp;</SPAN></P> </TD> <TD> <P><SPAN>- State store grows/shrinks based on the watermark delay.&nbsp;</SPAN></P> <P><SPAN>- For a watermark delay </SPAN><I><SPAN>D</SPAN></I><SPAN>, a record with event_time </SPAN><I><SPAN>T</SPAN></I><SPAN> gets removed from the state store only when the </SPAN><I><SPAN>watermark</SPAN></I><SPAN> is greater than or equal to </SPAN><I><SPAN>T</SPAN></I><SPAN> + </SPAN><I><SPAN>D</SPAN></I><SPAN>.</SPAN></P> </TD> </TR> <TR> <TD> <P><I><SPAN>dropDuplicates</SPAN></I><SPAN> + </SPAN><I><SPAN>withWatermark</SPAN></I><SPAN>, </SPAN><I><SPAN>event_time</SPAN></I><SPAN> is one of the deduplication keys.&nbsp;</SPAN></P> </TD> <TD> <P><SPAN>- When you expect the stream to have duplicates with the same </SPAN><I><SPAN>event_time</SPAN></I></P> </TD> <TD> <P><SPAN>- State store grows/shrinks based on the watermark delay.&nbsp;</SPAN></P> <P><SPAN>- For a watermark delay </SPAN><I><SPAN>D</SPAN></I><SPAN>, a record with event_time </SPAN><I><SPAN>T</SPAN></I><SPAN> gets removed from the state store only when the </SPAN><I><SPAN>watermark</SPAN></I><SPAN> is greater than or equal to </SPAN><I><SPAN>T</SPAN></I><SPAN>.</SPAN></P> </TD> </TR> </TBODY> </TABLE> <P>&nbsp;</P> Sat, 11 Jan 2025 03:43:11 GMT MuraliTalluri 2025-01-11T03:43:11Z https://community.databricks.com/t5/technical-blog/deep-dive-streaming-deduplication/ba-p/105062 <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="streaming_deduplication_watermarking.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14011i06436D4C43D28AC5/image-size/large?v=v2&amp;px=999" role="button" title="streaming_deduplication_watermarking.png" alt="streaming_deduplication_watermarking.png" /></span></P> Sat, 11 Jan 2025 03:43:11 GMT https://community.databricks.com/t5/technical-blog/deep-dive-streaming-deduplication/ba-p/105062 MuraliTalluri 2025-01-11T03:43:11Z https://community.databricks.com/t5/technical-blog/deep-dive-streaming-deduplication/bc-p/109569#M464 <P>Hi <a href="https://community.databricks.com/t5/user/viewprofilepage/user-id/48887">@MuraliTalluri</a>,</P><P>Thank you for such a detailed article.</P><P>I am following <EM>dropDuplicatesWithinWatermark</EM> with the same steps as yours. The only difference is that I am using autoloader and reading CSV files as the source and writing data to a Delta table. My stream is using <EM>trigger(once=True)</EM>. I upload the source file and observe the behavior after triggering the code.</P><DIV><DIV>&nbsp;</DIV><DIV>While inserting the following record, I noticed that it got dropped due to deduplication, not because of the watermark, even though its event time is less than the watermark threshold (i.e., 10:01 &lt; 10:02).</DIV><DIV>&nbsp;</DIV></DIV><P><STRONG><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="p2.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/14729i0DAFC3A8A39FABA1/image-size/large?v=v2&amp;px=999" role="button" title="p2.png" alt="p2.png" /></span></STRONG></P><P>&nbsp;</P><P>I inserted another record mimicking the same scenario as the above record, and this time, it was dropped due to the watermark as expected.</P><P>Could you please guide me on why there is this inconsistency in behavior?</P><P>Thanks</P> Mon, 10 Feb 2025 06:13:42 GMT https://community.databricks.com/t5/technical-blog/deep-dive-streaming-deduplication/bc-p/109569#M464 Muhammad_Umer 2025-02-10T06:13:42Z https://community.databricks.com/t5/technical-blog/deep-dive-streaming-deduplication/bc-p/122443#M654 <P>The semantics of watermarking any records that are older than watermark threshold may or may not get processed. In your case, the first attempt may have processed the record even thought it is older than watermark threshold. Reference -&nbsp;<A href="https://spark.apache.org/docs/latest/streaming/apis-on-dataframes-and-datasets.html#semantic-guarantees-of-stream-stream-inner-joins-with-watermarking" target="_blank">https://spark.apache.org/docs/latest/streaming/apis-on-dataframes-and-datasets.html#semantic-guarantees-of-stream-stream-inner-joins-with-watermarking</A></P> Sat, 21 Jun 2025 20:10:31 GMT https://community.databricks.com/t5/technical-blog/deep-dive-streaming-deduplication/bc-p/122443#M654 Sai_Nandam 2025-06-21T20:10:31Z