The Data Engineering Patterns I Wish Someone Had Taught Me Earlier

Six patterns that make your Spark pipelines more reliable, portable, and easier to maintain

I have worked as a Data Engineer for several years and have made my share of mistakes along the way. However, with experience, I have learned a few key tricks that really help you build reliable, clean, and reproducible code. This article focuses on a few powerful concepts that you should master as you build a career as a data engineer.

1. Know Your Platform Utilities: dbutils vs. notebookutils

If you work in Databricks, you will constantly use dbutils. However, Fabric users may be more familiar with its cousin notebookutils. Both work like a Swiss army knife by helping you read files in your storage account, connect to hidden secrets, and execute notebooks. Below are some examples.

# List files from your Unity Catalog

# Databricks

dbutils.fs.ls("/Volumes/<catalog>/<schema>/<volume>/data/")

# Fabric

notebookutils.fs.ls("Files/data/")

# Retrieve a secret token

#Databricks

token = dbutils.secrets.get(scope="my-scope", key="db-token")           

#Fabric

token = notebookutils.credentials.getSecret(                       

    "https://my-vault.vault.azure.net/", "db-token")

# Run a child notebook

# Databricks

dbutils.notebook.run("child_notebook", timeout_seconds=60,         

    arguments={"env": "prod"})

# Fabric

notebookutils.notebook.run("child_notebook", timeout_seconds=60,

    arguments={"env": "prod"})

Widgets allow you to use parameters in your code. If you are collaborating with a team, you can add a dropdown to force users to pick only valid parameters.

Databricks:

# Create widgets

dbutils.widgets.text("env", "dev", "Environment")

dbutils.widgets.dropdown("catalog", "bronze", ["bronze", "silver", "gold"], "Catalog")

# Read values

env     = dbutils.widgets.get("env")

catalog = dbutils.widgets.get("catalog")

Microsoft Fabric:  Fabric does not support notebookutils.notebook.widgets in the same way. Instead, parameters are passed to notebooks using the arguments dictionary in notebookutils.notebook.run(), or by defining a parameters cell at the top of the notebook and toggling it as a parameter cell in the cell toolbar.

2. Parameterize Your SQL with format()

Hard-coding catalog or table names into SQL strings creates brittle, environment-specific code. Use Python's format() method to inject parameters cleanly. In the example below, we are parameterizing our catalogs as dev_bronze and dev_silver. When promoting this code to a test environment, we can swap those values without touching a single line of SQL.

Start with a SQL file that uses placeholders:

SELECT o.order_id, o.order_date, c.customer_name

FROM {BRONZE_CATALOG}.raw.orders o

INNER JOIN {BRONZE_CATALOG}.raw.customers c

ON o.customer_id = c.customer_id

Then run this Python script to inject the values at runtime:

# Set variables — you can also retrieve these from widgets

BRONZE_CATALOG = "dev_bronze"

SILVER_CATALOG = "dev_silver"

def run_sql(sql_file: str, dst_table: str):

    with open(sql_file, "r") as f:

        sql = f.read().format(BRONZE_CATALOG=BRONZE_CATALOG,

                              SILVER_CATALOG=SILVER_CATALOG)

    spark.sql(f"CREATE OR REPLACE TABLE {SILVER_CATALOG}.dw.{dst_table} AS {sql}")

The final executed SQL will look like this:

CREATE OR REPLACE TABLE dev_silver.dw.fact_customer AS

SELECT o.order_id, o.order_date, c.customer_name

FROM dev_bronze.raw.orders o

INNER JOIN dev_bronze.raw.customers c

ON o.customer_id = c.customer_id

This pattern makes your pipelines portable across dev, test, and prod environments.  You can also use widgets to replace the variables above.

3. Speed Up Independent Tasks with Thread Pooling

When processing multiple independent units of work such as loading files, calling APIs, or running notebook jobs, running them sequentially wastes time. ThreadPoolExecutor lets you run them in parallel with minimal code.

from concurrent.futures import ThreadPoolExecutor

# 1. Define the list of notebooks you want to run simultaneously

notebooks_to_run = ["Extract_Sales", "Extract_Inventory", "Extract_Customers"]

# 2. Define the execution function

def run_notebook(notebook_name: str):

    print(f"🚀 Starting notebook: {notebook_name}")

    # Fabric syntax (Change to dbutils.notebook.run for Databricks)

    result = notebookutils.notebook.run(notebook_name, timeout_seconds=300)

    return f"✅ {notebook_name} finished. Result: {result}"

# 3. Run all notebooks in parallel using 3 worker threads

with ThreadPoolExecutor(max_workers=3) as pool:

    # pool.map automatically distributes the notebooks to the threads

    final_results = list(pool.map(run_notebook, notebooks_to_run))

# 4. Print the status report once all notebooks finish

print("\n--- Execution Summary ---")

for status in final_results:

    print(status)

Notice that we set max_workers to 3. You can scale this up or down to fit your needs. However, too many threads can overwhelm your cluster or downstream systems. It is also worth noting that this will not help with normal spark transformations since those are already run in parallel.  Ideally, this will work for APIs, running notebooks, and saving files.

4. Handle Schema Changes with mergeSchema

We have all had pipelines break because someone added a new column without letting anyone know. I have worked with the Facebook API, where Facebook would rename columns, add new ones, or simply delete columns you once expected. Instead of getting alerted by an error in the middle of the night and spending hours decoding someone else's code, use mergeSchema to handle schema evolution automatically.

df_new.write \

    .format("delta") \

    .mode("append") \

    .option("mergeSchema", "true") \

    .saveAsTable("silver.dw.customers")

This tells Delta Lake to extend the existing table schema when new columns appear, without dropping existing data. It is one of the simplest ways to make your pipelines resilient to schema drift.

5. Compare DataFrames with MD5 Hashing

Missing data between stages in your medallion architecture can be devastating. The business loses confidence in your data quality, and you now have hours of work to isolate and fix the problem. Comparing rows one by one is slow and error-prone. Instead, hash each row into a single fingerprint and compare hashes.

from pyspark.sql.functions import md5, concat_ws, col, coalesce, lit

def add_row_hash(df, key_col: str, hash_col: str = "row_hash"):

    value_cols = [c for c in df.columns if c != key_col]

    # Explicitly replace NULLs with a placeholder

    safe_cols = [coalesce(col(c).cast("string"), lit("~")) for c in value_cols]

    return df.withColumn(hash_col, md5(concat_ws("||", *safe_cols)))

df_old = add_row_hash(spark.read.table("silver.dw.customers_v1"), "customer_id")

df_new = add_row_hash(spark.read.table("silver.dw.customers_v2"), "customer_id")

# Find changed rows and show results

changed = df_new.join(df_old, "customer_id").filter( df_new.row_hash != df_old.row_hash)

display(changed)

This technique is fast, scalable, and works well for reconciliation workflows. Catch the errors in development before your users ever see the new Power BI dashboard.

6. Build Data Quality Checks Into Your Pipeline

Bad data is unavoidable, and it is partly the reason we all have jobs. However, if you are not on top of it, the business team will quickly lose confidence in you and your team. Adding lightweight checks at each layer of your pipeline helps identify problems early.

The code below looks for null values, duplicate rows, and empty tables using the assert statement. As your pipeline grows, you can add more robust checks, such as constraining values within specific columns or identifying strings inside an integer column.

from pyspark.sql.functions import col, count, countDistinct, sum, when

def run_quality_checks(df, table_name: str):

    metrics = df.agg(

        count("*").alias("row_count"),

        sum(when(col("customer_id").isNull(), 1).otherwise(0)).alias("null_count"),

        (count("customer_id") - countDistinct("customer_id")).alias("duplicate_count")

    ).collect()[0]

    row_count       = metrics["row_count"]

    null_count      = metrics["null_count"] or 0

    duplicate_count = metrics["duplicate_count"] or 0

    print(f"[{table_name}] Rows: {row_count} | Nulls: {null_count} | Duplicates: {duplicate_count}")

    assert null_count == 0,      f"{table_name}: NULL customer_id found"

    assert duplicate_count == 0, f"{table_name}: Duplicate customer_id found"

    assert row_count > 0,        f"{table_name}: Table is empty"

run_quality_checks(df_customers, "silver.dw.customers")

Notice how the assert statement is used to check your results.  These checks take seconds to write and can save hours of debugging. Consider adding them at every level of your medallion architecture. In a production environment, pair them with alerts to warn you of anomalies before they become incidents.

The difference between a good data engineer and a great one often comes down to the habits they build early. Parameterize your SQL, validate your data, plan for schema changes, and know your platform tools. These are not advanced concepts, but they are fundamentals that compound over time. The engineer who writes reliable, reproducible pipelines is the one teams trust and it also makes your job a lot more enjoyable and rewarding.