I have multiple transactional sources feeding into my Azure Databricks env (MySQL, MSSQL, MySQL datadumps) for a client company-wide DataLake. They are basically all "managed sources" (using different DBMSs, receiving application dumps, etc), but I don't trust that they're protected against schema drift and I want to ingest/store all raw data (because of my lack of faith in the sources, some of which are dependant on external third parties) in a batch ETL process, but process only the relevant data. Ideally, I don't want to store a copy of each source table for each ingestion. 

Because of the diversity of the data, I don't want to capture change for them all in the same way, so my solution is to keep a log of differences between the current state of the source table and a mirror of the previous ingest. I can then work on choosing how to capture the change for silver/gold on a case by case basis, by propagating changes from the logs into the clean tables.

I've got the current DLT solution below, but I suspect it's very inefficient and that there's a better way of doing things to leverage spark/Databricks functionality. I am also aware that I can easily run into an OOM or pagination issue because I'm loading whole tables, not streaming/batchign, and I'm not entirely sure how the DLT table will handle a change in the number of columns.

Cost is a signficant factor (client works mainly on donor funding). I've got a Unity Catalog-enabled ws, but am using core DLT.  i'm also concerned about the persistance of the DLT log tables. I'm not even sure if this is a reasonable use case for DLT. Data quantities are actually pretty small (each source on the order of GBs), but there are one or two monster tables.

Any suggestions for alternative more efficient solutions are welcome!
These are some of the solutions/threads I've been looking at:

Data change feed: https://docs.databricks.com/delta/delta-change-data-feed.html
https://community.databricks.com/t5/data-engineering/what-are-the-best-practices-for-change-data-cap...
https://community.databricks.com/t5/data-engineering/when-should-change-data-feed-be-used/td-p/26000

External management: https://community.databricks.com/t5/data-engineering/how-do-you-capture-change-logs-from-rdms-source...
https://www.databricks.com/blog/2019/07/15/migrating-transactional-data-to-a-delta-lake-using-aws-dm...
https://www.databricks.com/blog/2018/10/29/simplifying-change-data-capture-with-databricks-delta.htm...

DLT CDC: https://docs.gcp.databricks.com/delta-live-tables/cdc.html

similar discussions re ETL from transactional sources: 
https://community.databricks.com/t5/data-engineering/implement-autoloader-to-ingest-data-into-delta-...
https://stackoverflow.com/questions/59591536/how-to-compare-two-versions-of-delta-table-to-get-chang...

This is the actual function

from typing import List, Optional, Dict, Tuple, Set
from dataclasses import dataclass
from pandas import DataFrame as PandasDf
import dlt
from pyspark.sql import DataFrame as SparkDf
from pyspark.sql.column import Column
from pyspark.sql.functions import current_timestamp, lit

@dataclass
class Source:
    name: str
    server_url: str
    user: str
    password: str
    driver: str = "org.mariadb.jdbc.Driver"

def generate_diff_table(
    query: str, 
    table_name: str,
    mirror_table_location: str,
    diff_log_location: str,
    )-> None:
    
  @dlt.table(
    name= f"{table_name}_diff_log",
    comment=f"Log of changes to query: {query}",
    path=diff_log_location
  )
  def create_diff_table():
    timestamp: Column = current_timestamp()
    print(query)
    source_df: SparkDf = spark.read.format('jdbc').options(
            driver=source.driver,
            url=f"{source.server_url}",
            query=query,
            user=source.user,
            password=source.password,
            useSSL=True,
            trustServerCertificate=True,
            ).load()
    mirror_path: str = f"{mirror_table_location}.{table_name}_mirror"

# if the table has been ingested before, do set subtraction to get the differences between the current version and the previous version and store the differences as a new table (diff_df)
    if spark.catalog.tableExists(mirror_path):
        prev_mirror_df: SparkDf = spark.sql(f"Select * from {mirror_path}")
        diff_insert: SparkDf = source_df.subtract(prev_mirror_df).withColumn(colName="action", col=lit("insert")).withColumn(colName="ingest_timestamp", col=lit(timestamp))
        diff_del: SparkDf = prev_mirror_df.subtract(source_df).withColumn(colName="action", col=lit("delete")).withColumn(colName="ingest_timestamp", col=lit(timestamp))
        diff_df: SparkDf = diff_insert.union(diff_del)
    else:  # if the tbale has not been ingested before, all rows are essentially inserted
        diff_df = source_df.withColumn(colName="action", col=lit("insert")).withColumn(colName="ingest_timestamp", col=lit(timestamp))
    
    source_df.write.mode("overwrite").saveAsTable(mirror_path)
    return diff_df

-------------------------------------------------------