OK. So I think I'm probably missing the obvious and tying myself in knots here.
Here is the scenario:
- batch datasets arrive in json format in an Azure data lake
- each batch is a complete set of "current" records (the complete table)
- these are processed using autoloader in a streaming pipeline (because we have other related processes running streaming data pipelines and because autoloader makes life easy)
- duplicated data is removed from extracted rows
- changes are captured via built-in DLT SCD2 processing
This all working nicely.
Now, I need to identify when a record has been removed (deleted) and no longer appears in the batches.
In the SCD table, this should result in the status of that record changing from "current" to "former"
I am not managing to achieve this.
Here is the relevant pipeline code:
#autoloader picks up files
import dlt
@dlt.table
def currStudents_streamFiles():
return (
spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "json")
.option("cloudFiles.inferColumnTypes", "true")
.load("abfss://*****@********.dfs.core.windows.net/current-students/restAPI")
.select("*","_metadata", "_metadata.file_modification_time")
)
-- extract the rows from the json (each file is a complete dataset with a few hundred rows)
CREATE OR REFRESH STREAMING LIVE TABLE currStudents_ingest
AS SELECT
col.*
,"current" AS status
,file_modification_time
FROM (
SELECT fi.file_modification_time, EXPLODE_OUTER (fi.students)
FROM STREAM(LIVE.currStudents_streamFiles) AS fi
)
WHERE col.id IS NOT NULL
;
#run some de-duplication due to most records being identical each time
import dlt
from pyspark.sql.functions import lit
@dlt.table
def currStudents_dedup():
df = spark.readStream.format("delta").table("live.currStudents_ingest")
return (
df.dropDuplicates([col for col in df.columns if col != "file_modification_time"])
.select('*')
.withColumn('status', lit('current'))
)
-- capture SCD2 change history
CREATE OR REFRESH STREAMING LIVE TABLE students_SCD;
APPLY CHANGES INTO
live.currStudents_SCD
FROM
STREAM(live.currStudents_dedup)
KEYS
(id)
SEQUENCE BY
file_modification_time
STORED AS
SCD TYPE 2
TRACK HISTORY ON * EXCEPT (file_modification_time)
-- match the latest batch (from json file) against "current" version and identify missing records
-- attempt to push identified records back through SCD with a new status "former"
-- DLT pipeline doesn't like INSERT in the apply changes into ....
CREATE TEMPORARY LIVE VIEW former_students_view AS
SELECT *, "former" AS status
-- all records from the last batch processed
FROM (
SELECT *
FROM STREAM(live.currStudents_ingest)
WHERE file_modification_time = (
SELECT MAX(file_modification_time) FROM STREAM(live.currstudents_streamfiles)
)
) t1
WHERE NOT EXISTS (
-- "current" version of the table held in Databricks
SELECT 1
FROM (
SELECT schoolId FROM STREAM(live.students_SCD)
WHERE `__END_AT` IS NULL AND status != "former"
) t2
WHERE t1.schoolId = t2.schoolId
);
APPLY CHANGES INTO live.currStudents_dedup
FROM former_students_view
KEYS (schoolId)
INSERT ALL
All help gratefully received.
I am aware that I may be going about this in the wrong way.
