@Kearon McNicol​ :

In SCD Type 2, new records are created when there are changes to monitored columns as well as when there are no changes to monitored columns but the current record has expired. This is because the current record is considered outdated and a new record with the same key but updated values needs to be created. It's possible that you are seeing duplicate records because the current records have expired and new records are being created with the same key but the same values as the expired records.

To prevent this, you can try increasing the validity period of the records by setting a larger value for the valid_time_column in the STORED AS SCD 2 option. This will allow the records to remain valid for a longer period of time and reduce the number of new records that are created.

Alternatively, you can try using a deduplication step in your pipeline to remove the duplicate records before they are processed by the SCD Type 2 logic. This can be done by grouping the records by their key and selecting the most recent record for each group based on the file_modification_time column. This can be done using Spark's window function.

For example:

import pyspark.sql.functions as F
from pyspark.sql.window import Window
 
deduped_stream = (stream(live.currStudents_ingest)
                 .groupBy('id')
                 .agg(F.max('file_modification_time').alias('latest_file_modification_time'),
                      F.first('json', True).alias('json'))
                 .select('id', 'latest_file_modification_time', 'json'))
 
APPLY CHANGES INTO
  live.currStudents_SCD
FROM
  deduped_stream
KEYS
  (id)
SEQUENCE BY
  latest_file_modification_time
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT (latest_file_modification_time)

Here, we group the records by the id column and select the most recent record for each group based on the file_modification_time column. We then pass this deduplicated stream into the SCD Type 2 logic for further processing.

Hope this helps!