single parameterized notebook for all layer-wise row count and hash reconciliation

Question

single parameterized notebook for all layer-wise row count and hash reconciliation

Janice Chi 160

Full Question:

In our Azure-based data migration project, we are implementing row count and hash reconciliation between different storage layers across two phases:

One-time migration (DB2 Snapshot → ADLS CSV/Parquet → Azure SQL Hyperscale)

Catch-up CDC pipeline (Flattened CDC Table → ADLS Delta → Azure SQL Hyperscale)

To keep the solution modular and scalable, we want to avoid layer-specific notebooks like rowcount_compare_db2_adls, rowcount_compare_adls_sql, etc.

Instead, we plan to use only two generic, reusable notebooks:

rowcount_compare_generic

hash_compare_generic

These notebooks will be parameterized to receive:

source_layer_name and target_layer_name

table_name, partition_column, and partition_value

run_id and control_table_name

The notebook will:

Read the data from the source and target layers dynamically (DB2 via JDBC, ADLS via CSV/Parquet/Delta, Hyperscale via JDBC)

Perform either:

total row count comparison, or

  row-level hash comparison (for each partition or CDC offset window)
  
  Update the appropriate control/recon tables with match/mismatch status, retry_flag, etc.

We invoke these notebooks either from ADF (Execute Notebook activity) or from a Databricks driver notebook via dbutils.notebook.run().

Questions:

Does Microsoft recommend using such layer-agnostic, parameterized reconciliation notebooks?

Are there any performance, logging, or monitoring risks with dynamically switching data sources (JDBC, CSV, Parquet, Delta, etc.) within the same notebook?

Would Microsoft suggest any alternative design (e.g., shared wheels or functions) to improve reusability or maintainability further?

Venkat Reddy Navari 3,470 Reputation points Microsoft External Staff Moderator

2025-07-01T16:31:09.2633333+00:00

@Janice Chi We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.

1 answer

Your answer

Venkat Reddy Navari 3,470 Reputation points Microsoft External Staff Moderator

2025-07-01T16:31:09.2633333+00:00

@Janice Chi We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.

Answer 1

Venkat Reddy Navari 3,470 Microsoft External Staff Moderator

Hi @Janice Chi Designing reusable, parameterized notebooks for reconciliation is a smart way to simplify your data pipeline and make it easier to manage over time.

Is it a good idea to use generic, layer-agnostic reconciliation notebooks

Absolutely. In fact, this is a common and recommended approach, especially in enterprise-scale migrations or CDC flows like the one you’re working on.

Using notebooks like rowcount_compare_generic and hash_compare_generic helps you avoid repeating logic across layers and keeps things easier to update later. It’s a solid way to:

Standardize row-level checks across different systems
Reduce duplication in your Databricks or ADF environment
Enable metadata-driven orchestration (especially when you scale up to many tables)

This design is often paired with a control table that drives what to run and when — sounds like you’re already heading that way with parameters like run_id and control_table_name.

Are there any performance or reliability concerns when switching between source types (JDBC, CSV, Delta, etc.) dynamically

There can be, but they’re manageable if you plan for them. A few areas to watch:

Concern	Recommendation
JDBC connections	Use batching and proper connection handling. Connection pooling may help if needed.
JDBC connections	Use batching and proper connection handling. Connection pooling may help if needed.
Schema consistency	Add schema validation when reading from dynamic formats (CSV, Parquet, Delta).
Logging	Use lightweight logs (e.g., Delta tables, dbutils.fs logs) for partition/table-level visibility.
Error handling	Wrap logic in try/except blocks and push failures into the control table.
Monitoring	For ADF integration, surface notebook outputs and status codes for easier tracking.

Would share libraries or other design patterns help

Yes, and they’ll become more valuable as your project grows. Here are a few ideas:

Modular Python utilities Move shared read/compare logic to .py helper scripts or reusable notebooks (%run).
Custom wheel packages Package reconciliation logic as a Python wheel and install it on all Databricks clusters.
Control table–driven orchestration Let your pipeline logic read directly from a control/config table makes retry, skip, and audit much easier to manage.
Lightweight observability Log reconciliation results (row/hash match/mismatch, retry flags) to a Delta audit table. Visualize with Power BI or Grafana if needed.

I hope this information helps. Please do let us know if you have any further queries.

Kindly consider upvoting the comment if the information provided is helpful. This can assist other community members in resolving similar issues.

Janice Chi 160 Reputation points

2025-06-30T09:59:35.0466667+00:00

please help how we can set up framework and code everything for two generic, reusable notebooks:

rowcount_compare_generic

hash_compare_generic

These notebooks will be parameterized to receive:

source_layer_name and target_layer_name

table_name, partition_column, and partition_value

run_id and control_table_name

Venkat Reddy Navari 3,470 Microsoft External Staff Moderator

@Janice Chi You can implement a modular and metadata-driven reconciliation framework in Databricks by setting up:

Control Table (in Azure SQL or Delta)
Two Parameterized Notebooks:
- rowcount_compare_generic
- hash_compare_generic
Helper Modules / Utilities
Orchestration Layer (ADF or Driver Notebook)

CONTROL TABLE SCHEMA (Example)

Create a control table (SQL or Delta) to track reconciliation metadata:

sql
Copy
CREATE TABLE recon_control_table (
    run_id STRING,
    source_layer_name STRING,
    target_layer_name STRING,
    table_name STRING,
    partition_column STRING,
    partition_value STRING,
    compare_type STRING, -- 'rowcount' or 'hash'
    status STRING, -- 'pending', 'success', 'mismatch', 'error'
    retry_flag BOOLEAN,
    error_message STRING,
    execution_timestamp TIMESTAMP
);

rowcount_compare_generic Notebook

Parameters

python
Copy
dbutils.widgets.text("source_layer_name", "")
dbutils.widgets.text("target_layer_name", "")
dbutils.widgets.text("table_name", "")
dbutils.widgets.text("partition_column", "")
dbutils.widgets.text("partition_value", "")
dbutils.widgets.text("run_id", "")
dbutils.widgets.text("control_table_name", "")

Sample Logic

python
Copy
# Read parameters
source_layer = dbutils.widgets.get("source_layer_name")
target_layer = dbutils.widgets.get("target_layer_name")
table = dbutils.widgets.get("table_name")
partition_col = dbutils.widgets.get("partition_column")
partition_val = dbutils.widgets.get("partition_value")
run_id = dbutils.widgets.get("run_id")
control_table = dbutils.widgets.get("control_table_name")

from recon_utils import get_df

try:
    df_source = get_df(source_layer, table, partition_col, partition_val)
    df_target = get_df(target_layer, table, partition_col, partition_val)
    
    row_count_match = df_source.count() == df_target.count()
    
    status = "success" if row_count_match else "mismatch"
    error_msg = "" if row_count_match else "Row count mismatch"

except Exception as e:
    status = "error"
    error_msg = str(e)

# Write results back to control table
spark.sql(f"""
MERGE INTO {control_table} t
USING (SELECT '{run_id}' AS run_id) s
ON t.run_id = s.run_id AND t.table_name = '{table}' 
WHEN MATCHED THEN UPDATE SET 
    status = '{status}', 
    error_message = '{error_msg}', 
    execution_timestamp = current_timestamp()
""")

hash_compare_generic Notebook

Use the same structure, but instead of comparing counts, compute row-level hash:

Hash Logic Example:


from pyspark.sql.functions import sha2, concat_ws

def add_hash_column(df):
    return df.withColumn("row_hash", sha2(concat_ws("||", *df.columns), 256))

df_source = add_hash_column(df_source)
df_target = add_hash_column(df_target)

src_hashes = df_source.select("row_hash").distinct()
tgt_hashes = df_target.select("row_hash").distinct()

hash_mismatch = src_hashes.subtract(tgt_hashes).count() > 0 or \
                tgt_hashes.subtract(src_hashes).count() > 0

Then log match/mismatch in the same way as row count.

recon_utils.py Helper Module (reusable logic)


def get_df(layer, table_name, partition_col=None, partition_val=None):
    if layer == "db2":
        return spark.read \
            .format("jdbc") \
            .option("url", "<db2-url>") \
            .option("dbtable", table_name) \
            .option("user", "<user>") \
            .option("password", "<password>") \
            .load()
        
    elif layer == "adls_csv":
        path = f"abfss://<container>@<storage>.dfs.core.windows.net/{table_name}/*.csv"
        return spark.read.csv(path, header=True)
    
    elif layer == "adls_parquet":
        path = f"abfss://<container>@<storage>.dfs.core.windows.net/{table_name}/*.parquet"
        return spark.read.parquet(path)

    elif layer == "adls_delta":
        path = f"abfss://<container>@<storage>.dfs.core.windows.net/{table_name}"
        return spark.read.format("delta").load(path)
    
    elif layer == "sql_hyperscale":
        return spark.read \
            .format("jdbc") \
            .option("url", "<sql-hyperscale-url>") \
            .option("dbtable", table_name) \
            .option("user", "<user>") \
            .option("password", "<password>") \
            .load()
    
    else:
        raise ValueError(f"Unsupported layer: {layer}")

Orchestration via ADF or Driver Notebook

Loop through entries in the control table and invoke notebook:


notebook_path = "path/to/rowcount_compare_generic" if compare_type == "rowcount" else "hash_compare_generic"

dbutils.notebook.run(notebook_path, timeout_seconds=600, arguments={
    "source_layer_name": row.source_layer_name,
    "target_layer_name": row.target_layer_name,
    "table_name": row.table_name,
    "partition_column": row.partition_column,
    "partition_value": row.partition_value,
    "run_id": row.run_id,
    "control_table_name": control_table_name
})

I hope this information helps. Please do let us know if you have any further queries.

Kindly consider upvoting the comment if the information provided is helpful. This can assist other community members in resolving similar issues.

Share via

single parameterized notebook for all layer-wise row count and hash reconciliation

1 answer

Your answer