Lakeflow Declarative Pipelines – Delta Live Tables (DLT)

Delta Live Tables (DLT) is the former name of what Databricks now calls Lakeflow Declarative Pipelines.

In other words: Lakeflow Declarative Pipelines is the current evolution and product name of DLT.

Lakeflow Declarative Pipelines is a data-pipeline framework in Databricks for defining, executing, and maintaining data processing pipelines using a declarative programming model.

The key idea is:
You define the desired state and data relationships of your datasets, rather than explicitly controlling every execution step.
你描述“数据应该是什么样、数据之间是什么关系、数据质量应该满足什么条件”，而不是手工控制每一个执行步骤。

A Declarative Pipeline can contain:

data sources
data flows
streaming tables
materialized views
transformations
data-quality rules
dependencies between datasets
pipeline configuration
operational and monitoring information

The pipeline engine uses the definitions to construct a dependency graph / DAG and determine how the pipeline should be executed.

In current Databricks terminology, the technology formerly known as Delta Live Tables (DLT) has evolved into Lakeflow Declarative Pipelines. The underlying open framework is also referred to as Apache Spark Declarative Pipelines (SDP).

Our Data platform:

ADLS
↓
Bronze
↓
Silver
↓
Gold

Traditional data engineering flow looks this:

ADF Trigger
      ↓
Databricks Job
      ↓
Notebook 1
      ↓
Notebook 2
      ↓
Notebook 3
      ↓
Data Quality Job
      ↓
Aggregation Job
      ↓
Power BI Refresh

Data Engineer have to consider and orchestrate

When does it run?                         什么时候运行？
What are the dependencies?                谁依赖谁？
What is the execution order?              谁先执行？
How are failures handled?                 失败以后怎么办？
How do we handle re-runs / retries?       重新执行怎么办？
How is incremental processing handled?    如何处理增量？
How do we avoid duplicate data?           如何避免重复？
How is data quality enforced/validated?   如何做数据质量？
How is the pipeline monitored?            如何监控？

Declarative Pipeline does this:

Bronze = source data

Silver = cleaned Bronze data

Gold = business aggregation of Silver

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

Silver depends on Bronze

Gold depends on Silver

-----------------------
Bronze
   ↓
Silver
   ↓
Gold

This is called Dependency Graph.

The important architectural shift is: Execution logic becomes separated from data logic. 把 Data Logic 和 Execution Logic 分离。

Your code primarily describes:
(Data Engineers are more focused on: Data Engineer 更关注)

What data should exist? 数据是什么？
What transformation defines it? 数据怎么转换？
What quality rules must it satisfy? 数据质量要求是什么？
What datasets does it depend on? 数据依赖谁？

The platform can then manage much of (而 Pipeline Framework 更多负责):

Execution 怎么执行
Dependency resolution 依赖怎么解析
Incremental processing 怎么增量更新
Pipeline updates
Monitoring 怎么监控
Data quality metrics
Recovery 怎么处理失败
Parallel execution 哪些任务可以并行

                    Declarative Pipeline
                           │
          ┌────────────────┼────────────────┐
          │                │                │
          ↓                ↓                ↓
      Ingestion       Transformation    Data Quality
          │                │                │
     Auto Loader       Spark SQL       Expectations
          │                │                │
          ↓                ↓                ↓
       Bronze  ───────→  Silver  ─────→ Gold
          │
          ↓
      Delta Lake
          │
          ↓
   Unity Catalog

What exactly is the core problem that Declarative Pipelines (DP) aim to solve?

Assuming, a Bank, has those everyday:

Customer
Account
Transaction
Payment
Loan

They come from

Oracle
SQL Server
Kafka
ADLS
REST API

Data Engineer should create:

                Sources
                   │
       ┌───────────┼───────────┐
       ↓           ↓           ↓
    Oracle       Kafka       ADLS
       │           │           │
       └───────────┼───────────┘
                   ↓
                Bronze
                   ↓
                Silver
                   ↓
                 Gold
                   ↓
              Power BI

If we do not have a good pipeline framwork, data engineer may create

ADF Pipeline 1
ADF Pipeline 2
ADF Pipeline 3
Databricks Job 1
Databricks Job 2
Databricks Job 3
Notebook 1
Notebook 2
Notebook 3
DQ Job
CDC Job

maintaining

dependencies
schedules
retries
checkpoints
parameters
error handling
monitoring

One of the primary goals of Declarative Pipelines (DP) is to shift more of this complexity onto the pipeline framework.”

Key foundational concepts of Declarative Pipelines

Declarative Pipelines
│
├── Pipeline
│
├── Dataset
│   ├── Streaming Table
│   ├── Materialized View
│   └── Temporary View
│
├── Flow
│
├── Dependency Graph
│
├── Incremental Processing
│
├── Data Quality
│   └── Expectations
│
├── Change Data Capture
│   └── AUTO CDC
│
├── Batch Processing
│
├── Streaming Processing
│
├── Monitoring
│   └── Event Log
│
├── Unity Catalog
│
└── Deployment / CI-CD

Pipeline / Dataset / Flow

Concept	Professional Meaning	人话
Pipeline	Application / execution boundary	整个工厂
Dataset	Data object produced/maintained	产品
Flow	Data movement / processing path	传送带
Dependency	Relationship between datasets	谁必须等谁
Pipeline Engine	Executes the declared pipeline	工厂管理系统

Declarative Pipeline Python API code implement

# ============================================================
# Import the modern Declarative Pipelines API
# ============================================================

from pyspark import pipelines as dp


# ============================================================
# BRONZE DATASET
# ============================================================
#
# Purpose:
#   Define the Bronze dataset.
#
# Important:
#   We are DECLARING what this dataset should contain.
#   We are not manually starting a streaming query.
# ============================================================

@dp.table
def customer_bronze():

    return (
        spark.readStream

        # ----------------------------------------------------
        # Auto Loader
        #
        # Purpose:
        #   Incrementally discover new files arriving in ADLS.
        # ----------------------------------------------------
        .format("cloudFiles")

        # ----------------------------------------------------
        # Source format
        # ----------------------------------------------------
        .option("cloudFiles.format", "json")

        # ----------------------------------------------------
        # Source location
        # ----------------------------------------------------
        .load("/mnt/raw/customer")
    )


# ============================================================
# SILVER DATASET
# ============================================================
#
# Purpose:
#   Clean and standardize Bronze customer data.
#
# Dependency:
#
#       customer_bronze
#              ↓
#       customer_silver
#
# The dependency is expressed by reading customer_bronze.
# ============================================================

@dp.table
def customer_silver():

    return (
        spark.readStream

        # ----------------------------------------------------
        # Read the Bronze dataset.
        #
        # This establishes the logical dependency:
        #
        # customer_silver
        #        ↓
        # customer_bronze
        # ----------------------------------------------------
        .table("customer_bronze")

        # ----------------------------------------------------
        # Data quality / business rule:
        #
        # A customer record without customer_id is not useful
        # to downstream business processing.
        # ----------------------------------------------------
        .filter("customer_id IS NOT NULL")

        # ----------------------------------------------------
        # Keep only columns required by the Silver layer.
        # ----------------------------------------------------
        .select(
            "customer_id",
            "name",
            "email",
            "updated_at"
        )
    )

Step by Step build a Complete Lakeflow Declarative Pipeline

Customer Data Lakehouse Pipeline — Requirements

1. Project Goal / 项目目标

Build an end-to-end Lakeflow Declarative Pipeline for customer data:

ADLS Landing → Bronze → Data Quality / Routing → Silver → Gold

The pipeline must demonstrate:

Auto Loader incremental ingestion
Bronze raw-data preservation
Data Quality validation
Invalid-record quarantine
Severe-invalid record drop with audit logging
Silver trusted customer data
Gold business/analytics output
Declarative Pipeline dependency management
Incremental processing
Auditability and traceability

中文

我们建立一个完整的 Lakeflow Declarative Pipeline：

ADLS Landing → Bronze → 数据质量检查/路由 → Silver → Gold

这个实验不是简单地：

读数据 → filter → 写 Silver

而是模拟一个比较接近企业实际的 Data Engineering Pipeline。

重点学习：

Auto Loader 增量 ingestion
Bronze 保存原始数据
Data Quality validation
Bad data Quarantine
严重坏数据 Drop + Audit Log
Silver Trusted Data
Gold Analytics Data
Declarative Pipeline 自动 dependency management
Incremental processing
数据可追溯性和审计

2. Source Data / 数据源

Raw Data:
customer_001.json
customer_002.json
customer_003.json

ADLS Location: abfss://landing@.dfs.core.windows.net/customer/

ADLS
└── landing
    └── customer
        ├── customers_001.json
        ├── customers_002.json
        └── customers_003.json

Our pipeline will implement a simple Medallion Architecture using Lakeflow Declarative Pipelines.

                Azure ADLS
                    │
                    │ JSON files
                    ▼
              Auto Loader
                    │
                    ▼
          ┌───────────────────┐
          │ Bronze             │
          │ customer_bronze    │
          │ Raw / append       │
          └─────────┬─────────┘
                    │
                    │ transformation
                    ▼
          ┌───────────────────┐
          │ Silver            │
          │ customer_silver   │
          │ Cleaned data      │
          └─────────┬─────────┘
                    │
                    │ aggregation
                    ▼
          ┌───────────────────┐
          │ Gold              │
          │ customer_summary  │
          │ Business data     │
          └───────────────────┘

Customer Business Columns

Column	Meaning
customer_id	Customer identifier
name	Customer name
email	Customer email
status	ACTIVE / INACTIVE
country	Customer country
updated_at	Source/business update timestamp

3. Bronze Requirement / Bronze 要求

Bronze 是整个 pipeline 的原始数据入口层。Ingestion 必须使用：Auto Loader. 要能够：support scalable file ingestion
incremental discover new files
process new files
avoid reprocessing already processed files

4. Bronze Must Preserve Raw Data

这是非常重要的一条 requirement。Bronze 不能只保存清洗后的 business columns。Bronze 必须尽可能保留：

Business Data

customer_id
name
email
status
country
updated_at

Raw / Audit Metadata

例如：

_raw_record
_ingest_ts
_source_file
_source_file_name
_source_file_modification_time

如果适用，也可以考虑：

_source_system
_batch_id
_record_hash

5. Important Timestamp Definition

这里要特别区分两个时间, updated_at and _ingest_ts

updated_at: 这是 Source Business Timestamp, 表示源系统认为这条 customer record 什么时候更新。例如： updated_at = 2026-09-10 10:05:00

_ingest_ts: 这是 Data Platform Ingestion Timestamp, 表示这条数据什么时候进入我们的 Data Platform / Pipeline processing。例如：_ingest_ts = 2026-09-10 10:06:13

两者不能混为一谈。

6. Bronze Raw Record Requirement

我们希望 Bronze 能够保存原始 record 的 JSON representation，例如：_raw_record

这样以后：如果一条 record 被 quarantine 或 drop，我们可以看到它进入 pipeline 时的原始状态。这是企业 Data Platform 中非常重要的：Traceability / 可追溯性

7. Data Quality Requirements

这是整个实验最重要的部分。我们定义三个核心 Data Quality rules：

Rule 1 — Customer ID

customer_id must not be NULL

Rule 2 — Email

Email 必须：

NOT NULL

并且应该满足基本 email format。例如：alice@example.com 有效。而：alice@ 无效。

Rule 3 — Status

Status 必须是：ACTIVE 或者：INACTIVE . 其他值都是 invalid。例如：

ACTIVE       → valid
INACTIVE     → valid
PENDING      → invalid
UNKNOWN      → invalid
NULL         → invalid

8. Data Quality Routing Requirement

我们不是简单地 DROP 所有坏数据。这是这个项目的核心设计。我们把 record 分成三个类别：

                    Bronze
                       │
                       ▼
                Data Quality
                   Validation
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
        VALID      RECOVERABLE    SEVERE
          │            │            │
          ▼            ▼            ▼
       Silver      Quarantine    Drop Log

9. Routing Decision Matrix

这是目前我们确认的正式 requirement baseline。

Customer_ID	Email	Status	Action	Destination
Valid	Valid	ACTIVE	ACCEPT	Silver
Valid	Valid	INACTIVE	ACCEPT	Silver
NULL	Valid	Valid	QUARANTINE	Quarantine
Valid	NULL	Valid	QUARANTINE	Quarantine
Valid	Valid	Invalid	QUARANTINE	Quarantine
NULL	NULL	Valid	QUARANTINE	Quarantine
NULL	Valid	Invalid	QUARANTINE	Quarantine
Valid	NULL	Invalid	QUARANTINE	Quarantine
NULL	NULL	Invalid	DROP	Drop Log

10. Quarantine Requirement

如果 record 不符合正常 Silver 数据质量要求，但是没有达到 Severe Drop 条件：Quarantine

例如：

customer_id = NULL
email = bob@example.com
status = ACTIVE

不能进入 Silver。但是：不能直接删除。应该进入：

customer_quarantine

11. Quarantine Must Preserve Original Record

Quarantine 至少应该包含：Original Business Data

customer_id
name
email
status
country
updated_at

Original Record

_raw_record

Source Information

_source_file
_source_file_name

Ingestion Information

_ingest_ts

Data Quality Information

_dq_customer_id_valid
_dq_email_valid
_dq_status_valid

以及：

_dq_reason

例如：

CUSTOMER_ID_NULL

或者：

EMAIL_NULL

或者：

INVALID_STATUS

多个问题：

CUSTOMER_ID_NULL;EMAIL_NULL

12. Quarantine Timestamp

需要记录：_quarantine_ts . 表示：这条 record 被判定并进入 quarantine 的时间。

13. Drop Requirement

最严重的情况：

customer_id = NULL
email       = NULL
status      = INVALID

这条 record：不进入 Silver, 不作为正常 Quarantine Record. 而是：DROP

但是：Drop ≠ 数据完全消失,

这是非常重要的概念。真正的设计是：

Bronze
   │
   └── severe invalid
             │
             ▼
        Drop Audit Log

所以我们实际上是：从 business processing path 中 drop，但从 audit trail 中保留。

14. Drop Log Requirement

建立：customer_drop_log

至少保存：

Original Data

customer_id
name
email
status
country
updated_at

Raw Record

_raw_record

Source Metadata

_source_file
_source_file_name
_source_file_modification_time

Ingestion Information

_ingest_ts

Drop Information

_drop_ts
_drop_reason

例如：

_drop_reason =
CUSTOMER_ID_NULL;EMAIL_NULL;INVALID_STATUS

这样以后可以回答：Why was this customer record dropped?

以及：What exactly was the original record?

15. Silver Requirement

Silver 是：Trusted / Validated Customer Data

只有：

customer_id valid
AND
email valid
AND
status ∈ ('ACTIVE', 'INACTIVE')

的 record 才能进入 Silver。因此：

Bronze
   │
   ├── valid ─────────────→ Silver
   │
   ├── invalid/recoverable → Quarantine
   │
   └── severe invalid ────→ Drop Log

Silver 不应该包含：

NULL customer_id
NULL email
invalid email
invalid status

16. Silver Transformation Requirement

Silver 除了 DQ filtering，还可以做一些正常的 standardization，例如：

trim(customer_id)
trim(name)
lower(email)
upper(status)

例如：

" alice@example.com "

standardize 成：

"alice@example.com"

但是：Bronze 不修改原始数据。这是：

Bronze = Preserve
Silver = Clean / Standardize / Trust

17. Gold Requirement

Gold 是面向：Business / Analytics Consumption. 我们需要一个简单但有意义的 Gold dataset。

例如：

customer_summary

按：

status

统计 customer 数量：

status      customer_count
---------------------------
ACTIVE      5
INACTIVE    1

也可以进一步按：

country
status

进行 aggregation。

例如：

country    status       customer_count
---------------------------------------
Canada     ACTIVE       4
Canada     INACTIVE     0
USA        INACTIVE     1
UK         ACTIVE       1

Gold 使用：Materialized View 来体现：

Silver
   ↓
Gold Materialized View

18. Final Pipeline Architecture

                         ADLS
                          │
                          │
                    Auto Loader
                          │
                          ▼
                    ┌───────────┐
                    │  Bronze   │
                    │ Raw Data  │
                    │ + Metadata│
                    └─────┬─────┘
                          │
                          ▼
                  Data Quality Rules
                          │
             ┌────────────┼────────────┐
             │            │            │
             ▼            ▼            ▼
          VALID      RECOVERABLE     SEVERE
             │            │            │
             ▼            ▼            ▼
          Silver      Quarantine    Drop Log
             │
             │
             ▼
       Gold Materialized
             View

19. Final Dataset List

整个 Pipeline 最终应该至少有 4 个主要数据集：

1. customer_bronze
2. customer_silver
3. customer_quarantine
4. customer_drop_log
5. customer_summary        ← Gold

也就是说：

Bronze
   │
   ├──────────────→ Silver
   │                   │
   │                   ▼
   │                  Gold
   │
   ├──────────────→ Quarantine
   │
   └──────────────→ Drop Log

20. Declarative Pipeline Requirement

我们不希望自己手工 orchestration, 而是 Declare the datasets and their relationships.

Silver reads Bronze
Gold reads Silver
Quarantine reads Bronze
Drop Log reads Bronze

Pipeline framework 根据这些 dependencies 自动建立：

Dependency Graph

然后负责：

execution
dependency resolution
orchestration
incremental processing
checkpointing
pipeline updates
table maintenance

我们的代码主要负责：定义“数据应该是什么样子，以及它从哪里来”。

22. Incremental Processing Requirement

实验必须体现：First Run 例如：customers_001.json ; customers_002.json, 进入 pipeline。

Later 再增加：customers_003.json

Pipeline update 后：只处理新的 input data，而不是每次重新从头读取所有文件。

最终我们观察：

Bronze
Silver
Quarantine
Drop Log
Gold

23. Data Lineage / Traceability Requirement

我们最终应该能够回答：Where did this Silver record come from?

Silver
 ↓
Bronze
 ↓
ADLS file

Why was this record quarantined?

Quarantine
 ↓
DQ reason
 ↓
Original record
 ↓
Source file

Why was this record dropped?

Drop Log
 ↓
Drop reason
 ↓
Original record
 ↓
Source file
 ↓
Ingestion timestamp

这就是：

Data Lineage + Auditability + Traceability

25. Coding Order / 我们之后的 Coding 顺序

Step 1: Create Bronze

ADLS
 ↓
Auto Loader
 ↓
customer_bronze

Step 2: Validate Bronze Data

建立：

customer_id validation
email validation
status validation

Step 3: Classify / Route Records

VALID
QUARANTINE
DROP

Step 4: Create Quarantine

customer_quarantine

Step 5: Create Drop Audit Log

customer_drop_log

Step 6: Create Silver

customer_silver

Step 7: Create Gold

customer_summary

Step 8: Test Incremental Processing

增加新的 JSON：

customers_003.json

Step 9: Review Pipeline Graph / Data Quality / Audit

Test Data

we want to demonstrate incremental ingestion.

# File 1 — customers_001.json

[
  {
    "customer_id": "C001",
    "name": "Alice",
    "email": "alice@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:00:00"
  },
  {
    "customer_id": "C002",
    "name": "Bob",
    "email": "bob@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:05:00"
  },
  {
    "customer_id": "C003",
    "name": "Charlie",
    "email": "charlie@example.com",
    "status": "INACTIVE",
    "country": "USA",
    "updated_at": "2026-09-10T09:10:00"
  }
]





# File 2 — customers_002.json
[
  {
    "customer_id": "C004",
    "name": "David",
    "email": "david@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T10:00:00"
  },
  {
    "customer_id": "C005",
    "name": "Emma",
    "email": "emma@example.com",
    "status": "ACTIVE",
    "country": "UK",
    "updated_at": "2026-09-10T10:05:00"
  }
]

URL: abfss://landing@storageaccount.dfs.core.windows.net/customer/

Pipeline source code will like:

customer_pipeline/
│
├── customer_pipeline.py
│
└── configuration

1: Create Bronze (Simple, no for product)

aim: Incrementally load JSON files from ADLS into a Bronze Streaming Table.

# 简单实现， 其实在实际项目不会这么简单
from pyspark import pipelines as dp


@dp.table 
def customer_bronze():        # create a table called "customer_bronze" 
   # The decorator declares that the function defines a pipeline-managed dataset.
   # In this context, it creates a streaming table.

    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("abfss://landing@myaccount.dfs.core.windows.net/customer/")
    )

we do not do ” create table ……” then “writeStream.start()” ,

Declarative Pipeline WILL TAKE OVER this job

The pipeline framework manages:

execution
checkpointing
dependency resolution
incremental processing
scheduling
table maintenance
orchestration

The developer primarily declares: what dataset should exist and how it is derived.

2: Create Silver (Simple, not for product)

@dp.table
@dp.expect(                          # Data Quality Expectation
    "valid_customer_id",
    "customer_id IS NOT NULL"        # customer_id cannot null
)

@dp.expect(
    "valid_email",
    "email IS NOT NULL AND email RLIKE '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'"
                                           # email cannot be null and it should be in good form
)

@dp.expect(
    "valid_status",
    "status IN ('ACTIVE', 'INACTIVE')"    # status only allow ACTIVE or INACTIVE
)

def customer_silver():

    return (
        spark.readStream
            .table("customer_bronze")
            .filter("customer_id IS NOT NULL")
            .select(
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at"
            )
    )

Product grade, end- to-end code implement

test data

abfss://landing@<storage-account>.dfs.core.windows.net/customer/

customers_001.json
customers_002.json

[
  {
    "customer_id": "C001",
    "name": "Alice",
    "email": "alice@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:00:00"
  },
  {
    "customer_id": "C002",
    "name": "Bob",
    "email": "bob@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:05:00"
  },
  {
    "customer_id": "C003",
    "name": "Charlie",
    "email": "charlie@example.com",
    "status": "INACTIVE",
    "country": "USA",
    "updated_at": "2026-09-10T09:10:00"
  }
]

Step 1 — Create Bronze

Our first pipeline layer is:

ADLS Landing
      │
      │ Auto Loader
      ▼
customer_bronze

Bronze’s job is not to clean the data. Its job is: Ingest and preserve the source data faithfully, together with ingestion/source metadata.

First Bronze implementation

from pyspark import pipelines as dp


@dp.table
def customer_bronze():  # create a table called "customer_bronze" 

    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load(
                "abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
            )
    )

What does this code actually produce? it creates “customer_bronze”

Table customer_bronze:

customer_id
name
email
status
country
updated_at

# we're missing
├── _raw_record
├── _ingest_ts
├── _source_file
├── _source_file_name
└── _source_file_modification_time
这些是我们之前 Requirement 中要求的 audit information。只有改成下面的

现在我们把代码升级，满足requirement 的要求。

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    current_timestamp,
    col,
    struct,
    to_json
)


@dp.table(
    comment="Bronze layer containing raw customer data ingested incrementally from ADLS using Auto Loader."
)
def customer_bronze():

    return (
        spark.readStream

            # ---------------------------------------------------------
            # Use Databricks Auto Loader as the streaming file source.
            # Auto Loader incrementally discovers new files arriving
            # in the ADLS landing folder.
            # ---------------------------------------------------------
            .format("cloudFiles")

            # ---------------------------------------------------------
            # The source files in our experiment are JSON files.
            # ---------------------------------------------------------
            .option("cloudFiles.format", "json")

            # ---------------------------------------------------------
            # Load files from the ADLS landing zone.
            #
            # Replace <storage-account> with the real storage account.
            # ---------------------------------------------------------
            .load(
                "abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
            )

            # ---------------------------------------------------------
            # Add platform ingestion timestamp.
            #
            # This is NOT the source business timestamp.
            # updated_at = source/business time
            # _ingest_ts = data-platform ingestion/processing time
            # ---------------------------------------------------------
            .withColumn(
                "_ingest_ts",
                current_timestamp()
            )

            # ---------------------------------------------------------
            # Capture source file metadata.
            #
            # _metadata is provided by Auto Loader.
            # We preserve the file path and modification time so that
            # downstream DQ issues can be traced back to the source file.
            # ---------------------------------------------------------
            .withColumn(
                "_source_file",
                col("_metadata.file_path")
            )
            .withColumn(
                "_source_file_modification_time",
                col("_metadata.file_modification_time")
            )

            # ---------------------------------------------------------
            # Keep the raw source record as JSON.
            #
            # This gives us a representation of the record as it exists
            # at the Bronze stage before Silver transformations.
            #
            # This will later be extremely useful for:
            #   - Quarantine
            #   - Drop Audit Log
            #   - Troubleshooting
            #   - Data lineage / traceability
            # ---------------------------------------------------------
            .withColumn(
                "_raw_record",
                to_json(
                    struct(
                        "customer_id",
                        "name",
                        "email",
                        "status",
                        "country",
                        "updated_at"
                    )
                )
            )

            # ---------------------------------------------------------
            # Select the columns that belong to the Bronze contract.
            #
            # Bronze preserves business data plus source/audit metadata.
            # ---------------------------------------------------------
            .select(
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",

                "_raw_record",
                "_ingest_ts",

                "_source_file",
                "_source_file_modification_time"
            )
    )

三个时间的区别 (以后你面试很可能会被问到)

Field	Meaning
`updated_at`	Source business timestamp
`_source_file_modification_time`	Source file timestamp
`_ingest_ts`	Platform ingestion timestamp

为什么需要 _raw_record？这是我们后面设计：

Quarantine
Drop Log

Current Bronze Architecture

现在我们已经从：

ADLS
 ↓
Auto Loader
 ↓
Bronze

升级成：

                         ADLS
                          │
                          ▼
                    Auto Loader
                          │
                          ▼
                ┌──────────────────┐
                │ customer_bronze  │
                ├──────────────────┤
                │ customer_id      │
                │ name             │
                │ email            │
                │ status           │
                │ country          │
                │ updated_at       │
                │                  │
                │ _raw_record      │
                │ _ingest_ts       │
                │ _source_file     │
                │ _source_file_... │
                └──────────────────┘

What should we see after the first run? 如果 source 中：

C001
C002
C003
C004
C005

那么 Bronze 应该有：5 records. 例如：

customer_id	email	status	_source_file	_ingest_ts
C001	alice@example.com	ACTIVE	customers_001.json	…
C002	bob@example.com	ACTIVE	customers_001.json	…
C003	charlie@example.com	INACTIVE	customers_001.json	…
C004	david@example.com	ACTIVE	customers_002.json	…
C005	emma@example.com	ACTIVE	customers_002.json	…

注意：我们还没有进行任何 Business Data Quality filtering。

所以把代码补完整：

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    current_timestamp,
    col,
    struct,
    to_json
)

@dp.table(
    comment="Bronze layer containing raw customer data ingested incrementally from ADLS using Auto Loader."
)
def customer_bronze():

    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load(
                "abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
            )

            # Platform ingestion timestamp
            .withColumn(
                "_ingest_ts",
                current_timestamp()
            )

            # Source file metadata
            .withColumn(
                "_source_file",
                col("_metadata.file_path")
            )
            .withColumn(
                "_source_file_name",
                col("_metadata.file_name")
            )
            .withColumn(
                "_source_file_modification_time",
                col("_metadata.file_modification_time")
            )

            # Serialized representation of the parsed business record
            .withColumn(
                "_raw_record",
                to_json(
                    struct(
                        "customer_id",
                        "name",
                        "email",
                        "status",
                        "country",
                        "updated_at"
                    )
                )
            )

            .select(
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",
                "_raw_record",
                "_ingest_ts",
                "_source_file",
                "_source_file_name",
                "_source_file_modification_time"
            )
    )

Step 2 — Data Quality Validation

数据质量验证

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    col,
    when,
    concat_ws,
    array,
    lit
)

@dp.temporary_view
def customer_validation():

    return (
        spark.readStream
            .table("customer_bronze")

            # ==============================
            # DQ Rule 1: Customer ID
            # ==============================
            .withColumn(
                "_dq_customer_id_valid",
                col("customer_id").isNotNull()
            )

            # ==============================
            # DQ Rule 2: Email
            # ==============================
            .withColumn(
                "_dq_email_valid",
                (
                    col("email").isNotNull()
                    &
                    col("email").rlike(
                        r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                    )
                )
            )

            # ==============================
            # DQ Rule 3: Status
            # ==============================
            .withColumn(
                "_dq_status_valid",
                col("status").isin("ACTIVE", "INACTIVE")
            )

            # ==============================
            # Overall validation
            # ==============================
            .withColumn(
                "_dq_all_valid",
                (
                    col("_dq_customer_id_valid")
                    &
                    col("_dq_email_valid")
                    &
                    col("_dq_status_valid")
                )
            )

            # ==============================
            # DQ reason
            # ==============================
            .withColumn(
                "_dq_reason",
                concat_ws(
                    ";",
                    array(
                        when(
                            col("customer_id").isNull(),
                            lit("CUSTOMER_ID_NULL")
                        ),

                        when(
                            col("email").isNull(),
                            lit("EMAIL_NULL")
                        ),

                        when(
                            col("email").isNotNull()
                            &
                            ~col("email").rlike(
                                r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                            ),
                            lit("INVALID_EMAIL_FORMAT")
                        ),

                        when(
                            ~col("status").isin("ACTIVE", "INACTIVE"),
                            lit("INVALID_STATUS")
                        )
                    )
                )
            )

            # ==============================
            # Routing decision
            # ==============================
            .withColumn(
                "_dq_action",
                when(
                    col("_dq_all_valid"),
                    lit("ACCEPT")
                )
                .when(
                    (~col("_dq_customer_id_valid"))
                    &
                    (~col("_dq_email_valid"))
                    &
                    (~col("_dq_status_valid")),
                    lit("DROP")
                )
                .otherwise(
                    lit("QUARANTINE")
                )
            )
    )

Step 3 — Routing into Silver / Quarantine / Drop Log

Silver 的原则：Only trusted, validated customer records enter Silver.

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    col,
    when,
    concat_ws,
    array,
    lit,
    trim,
    lower,
    upper,
    current_timestamp
)


# ============================================================
# 1. Validation and Routing
# ============================================================

@dp.temporary_view
def customer_validation():

    return (
        spark.readStream
            .table("customer_bronze")

            # ------------------------------------------------
            # DQ Rule 1: customer_id must not be NULL
            # ------------------------------------------------
            .withColumn(
                "_dq_customer_id_valid",
                col("customer_id").isNotNull()
            )

            # ------------------------------------------------
            # DQ Rule 2:
            # email must not be NULL and must match
            # the basic email format
            # ------------------------------------------------
            .withColumn(
                "_dq_email_valid",
                (
                    col("email").isNotNull()
                    &
                    col("email").rlike(
                        r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                    )
                )
            )

            # ------------------------------------------------
            # DQ Rule 3:
            # status must be ACTIVE or INACTIVE
            # ------------------------------------------------
            .withColumn(
                "_dq_status_valid",
                (
                    col("status").isNotNull()
                    &
                    col("status").isin(
                        "ACTIVE",
                        "INACTIVE"
                    )
                )
            )

            # ------------------------------------------------
            # Overall DQ result
            # ------------------------------------------------
            .withColumn(
                "_dq_all_valid",
                (
                    col("_dq_customer_id_valid")
                    &
                    col("_dq_email_valid")
                    &
                    col("_dq_status_valid")
                )
            )

            # ------------------------------------------------
            # DQ Reason
            #
            # Multiple reasons are concatenated:
            #
            # CUSTOMER_ID_NULL;EMAIL_NULL
            # ------------------------------------------------
            .withColumn(
                "_dq_reason",
                concat_ws(
                    ";",
                    array(

                        when(
                            col("customer_id").isNull(),
                            lit("CUSTOMER_ID_NULL")
                        ),

                        when(
                            col("email").isNull(),
                            lit("EMAIL_NULL")
                        ),

                        when(
                            col("email").isNotNull()
                            &
                            ~col("email").rlike(
                                r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                            ),
                            lit("INVALID_EMAIL_FORMAT")
                        ),

                        when(
                            col("status").isNull()
                            |
                            ~col("status").isin(
                                "ACTIVE",
                                "INACTIVE"
                            ),
                            lit("INVALID_STATUS")
                        )
                    )
                )
            )

            # ------------------------------------------------
            # Routing Decision
            #
            # ACCEPT:
            #   All three rules pass
            #
            # DROP:
            #   All three rules fail
            #
            # QUARANTINE:
            #   Everything else
            # ------------------------------------------------
            .withColumn(
                "_dq_action",
                when(
                    col("_dq_all_valid"),
                    lit("ACCEPT")
                )
                .when(
                    (~col("_dq_customer_id_valid"))
                    &
                    (~col("_dq_email_valid"))
                    &
                    (~col("_dq_status_valid")),
                    lit("DROP")
                )
                .otherwise(
                    lit("QUARANTINE")
                )
            )
    )


# ============================================================
# 2. Silver
# ============================================================

@dp.table(
    comment="Trusted customer records that passed all core data quality rules."
)
def customer_silver():

    return (
        spark.readStream
            .table("customer_validation")

            # Only fully valid records
            .filter(
                col("_dq_action") == "ACCEPT"
            )

            # Standardize trusted business data
            .select(
                trim(
                    col("customer_id")
                ).alias("customer_id"),

                trim(
                    col("name")
                ).alias("name"),

                lower(
                    trim(
                        col("email")
                    )
                ).alias("email"),

                upper(
                    trim(
                        col("status")
                    )
                ).alias("status"),

                trim(
                    col("country")
                ).alias("country"),

                col("updated_at"),

                # Audit / lineage metadata
                col("_raw_record"),
                col("_ingest_ts"),
                col("_source_file"),
                col("_source_file_name"),
                col("_source_file_modification_time")
            )
    )


# ============================================================
# 3. Quarantine
# ============================================================

@dp.table(
    comment="Customer records that failed one or more recoverable data quality rules."
)
def customer_quarantine():

    return (
        spark.readStream
            .table("customer_validation")

            # Recoverable DQ failures
            .filter(
                col("_dq_action") == "QUARANTINE"
            )

            .select(
                # Original business data
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",

                # Raw / lineage information
                "_raw_record",
                "_ingest_ts",
                "_source_file",
                "_source_file_name",
                "_source_file_modification_time",

                # DQ results
                "_dq_customer_id_valid",
                "_dq_email_valid",
                "_dq_status_valid",
                "_dq_reason",

                # Quarantine audit timestamp
                current_timestamp().alias(
                    "_quarantine_ts"
                )
            )
    )


# ============================================================
# 4. Drop Log
# ============================================================

@dp.table(
    comment="Audit log for severely invalid customer records excluded from normal processing."
)
def customer_drop_log():

    return (
        spark.readStream
            .table("customer_validation")

            # Severe DQ failures
            .filter(
                col("_dq_action") == "DROP"
            )

            .select(
                # Original business data
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",

                # Raw / lineage information
                "_raw_record",
                "_ingest_ts",
                "_source_file",
                "_source_file_name",
                "_source_file_modification_time",

                # DQ results
                "_dq_customer_id_valid",
                "_dq_email_valid",
                "_dq_status_valid",

                # Audit reason
                col("_dq_reason").alias(
                    "_drop_reason"
                ),

                # Drop audit timestamp
                current_timestamp().alias(
                    "_drop_ts"
                )
            )
    )

这四个 Dataset 的关系

                    customer_bronze
                           │
                           │
                           ▼
                ┌────────────────────┐
                │ customer_validation│
                │                    │
                │ DQ Rules           │
                │ + Reason           │
                │ + Routing          │
                └─────────┬──────────┘
                          │
             ┌────────────┼────────────┐
             │            │            │
          ACCEPT      QUARANTINE      DROP
             │            │            │
             ▼            ▼            ▼
     customer_silver   customer_     customer_
                      quarantine    drop_log

最终 Routing Logic

customer_id	email	status	`_dq_action`	去哪里
valid	valid	ACTIVE	ACCEPT	Silver
valid	valid	INACTIVE	ACCEPT	Silver
NULL	valid	valid	QUARANTINE	Quarantine
valid	NULL	valid	QUARANTINE	Quarantine
valid	invalid format	valid	QUARANTINE	Quarantine
valid	valid	invalid	QUARANTINE	Quarantine
NULL	NULL	valid	QUARANTINE	Quarantine
NULL	valid	invalid	QUARANTINE	Quarantine
valid	NULL	invalid	QUARANTINE	Quarantine
NULL	NULL	invalid	DROP	Drop Log

Step 4 — Gold: customer_summary

from pyspark import pipelines as dp


@dp.materialized_view(
    comment="Gold customer summary by customer status."
)
def customer_summary():

    return (
        spark.read
            .table("customer_silver")
            .groupBy("status")
            .count()
            .withColumnRenamed(
                "count",
                "customer_count"
            )
    )

完整架构：

                       ADLS Landing
                            │
                            ▼
                     ┌─────────────┐
                     │ Auto Loader │
                     └──────┬──────┘
                            │
                            ▼
                    ┌───────────────┐
                    │    Bronze     │
                    │ Raw + Metadata│
                    └───────┬───────┘
                            │
                            ▼
                 ┌─────────────────────┐
                 │ Customer Validation │
                 │                     │
                 │ DQ Rules            │
                 │ DQ Reason           │
                 │ Routing             │
                 └─────────┬───────────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
           ACCEPT      QUARANTINE      DROP
              │            │            │
              ▼            ▼            ▼
           Silver      Quarantine    Drop Log
              │
              ▼
        Materialized View
              │
              ▼
             Gold
              │
              ▼
     customer_summary

Lakeflow Declarative Pipeline

我们现在写的是：

@dp.table
def customer_bronze():
    ...

然后：

@dp.temporary_view
def customer_validation():
    return spark.readStream.table("customer_bronze")

然后：

@dp.table
def customer_silver():
    return spark.readStream.table("customer_validation")

最后：

@dp.materialized_view
def customer_summary():
    return spark.read.table("customer_silver")

Streaming Table vs Materialized View

	Streaming Table	Materialized View
典型用途	incremental row processing	derived / aggregated result
数据特点	持续到来的数据	查询计算结果
常见操作	filter / transform / enrich	aggregation / business metrics
API	`@dp.table`	`@dp.materialized_view`
我们的例子	`customer_silver`	`customer_summary`

Step 6 — Put the Code into a Lakeflow Pipeline

现在进入真正的 hands-on 部分：

我们已经写完 Dataset definitions。下一步就是把这些 definitions 放进一个 Lakeflow Pipeline，让 Databricks 真正执行它。

这里先只做 Pipeline structure + code organization，暂时不深入 Deployment / Schedule / Production settings。

先把代码组织起来

最简单的方式是一个 Python pipeline source file，例如：

customer_pipeline/
│
└── customer_pipeline.py

里面包含我们目前的全部 Dataset definitions：

customer_pipeline.py
│
├── customer_bronze
│
├── customer_validation
│
├── customer_silver
│
├── customer_quarantine
│
├── customer_drop_log
│
└── customer_summary

Data Federation

Data Federation lets Databricks/Unity Catalog access and query data that remains in its original system, without first copying all of that data into Delta tables in Databricks.

Suppose I have multiple data:
on-prime: SQL Server, MySQL, Oracle;
Clould Azure SQL DB;
Cloud files on ADLS, format in Delta, Clould files on Azure Blob

My target environment is:

                    Azure Databricks
                         │
                  Unity Catalog
                         │
              ┌──────────┴──────────┐
              │   Data Federation   │
              └──────────┬──────────┘
                         │
       ┌─────────────────┼─────────────────┐
       │                 │                 │
   On-Prem             Azure             Cloud Files
       │                 │                 │
 ┌─────┼─────┐       Azure SQL DB      ADLS Gen2
 │     │     │                         Delta files
SQL  MySQL Oracle                      Blob Storage
Server

Think of two fundamentally different approaches. Traditional approach — copy the data
SQL Server
│
│ ETL / ADF / Spark
▼
ADLS
│
▼
Delta Lake
│
▼
Unity Catalog
│
▼
Databricks SQL

The data physically moves.

Federation approach, the customer data can remain inside SQL Server.

                    Unity Catalog
                         │
                 Foreign Catalog
                         │
        ┌────────────────┼────────────────┐
        │                │                │
    SQL Server        Oracle          MySQL
        │                │                │
        └────────────────┼────────────────┘
                         │
                  Query Federation

The data stays where it is. You can query data where it lives.

Databricks provides a governed way to access it.

SELECT *
FROM sqlserver_catalog.sales.customers;

The customer data can remain inside SQL Server.

The Unity Catalog mental model

Federation through this hierarchy:

Unity Catalog
│
├── Metastore
│
├── Catalog
│
│   ├── Schema
│   │
│   │   └── Table
│   │
│   └── ...
│
├── Connections
│
├── Foreign Catalogs
│
├── Storage Credentials
│
├── External Locations
│
└── Volumes

For Federation, pay particular attention to:

Connection
    ↓
Foreign Catalog
    ↓
Foreign Schema
    ↓
Foreign Table

Conceptually

SQL Server
    │
    │ connection
    ▼
Unity Catalog
    │
    ▼
Foreign Catalog
    │
    ├── schema1
    │     ├── tableA
    │     └── tableB
    │
    └── schema2
          └── tableC

Hands-on learning lab

We’ll create:

LAB
│
├── SQL Server
│
├── MySQL
│
├── Oracle
│
├── Azure SQL Database
│
├── ADLS Gen2
│   └── Delta
│
└── Azure Blob
    └── CSV/Parquet

And connect these to:

Azure Databricks
       │
       ▼
Unity Catalog
       │
       ├── sqlserver_federated
       ├── mysql_federated
       ├── oracle_federated
       ├── azuresql_federated
       │
       ├── catalog_ADLS_Internal
       │
       └── blob_external

Then we’ll run queries combining the data.

Lab scenario

Let’s create a realistic business scenario. Suppose the company has:

SQL Server

Customer master:Customer

CustomerID
CustomerName
Region
CustomerType

Oracle

Orders:Orders

OrderID
CustomerID
OrderDate
Amount

MySQL

Products:

ProductID
ProductName
Category
Price

Azure SQL

Customer transactions:

Transaction

TransactionID
CustomerID
ProductID
Amount
TransactionDate

ADLS Delta

Historical sales: sales_history

Blob

Marketing CSV: customer_campaign.csv

Now we can ask: Show total customer sales across Oracle historical orders, Azure SQL transactions and ADLS Delta, enriched with customer information from SQL Server.

Step 1 — Prepare Unity Catalog

First verify your current UC environment, ensure we have enable “Unity Catalog”, and you have grant to create catalog, schema, table, external location, Connection

Step 2 — Understand the Connection

This is one of the most important concepts. A Connection represents how Databricks reaches the external database.

# Conceptually:

Unity Catalog
     │
     ▼
Connection
     │
     ├── hostname
     ├── port
     ├── database
     ├── authentication
     └── credentials
          │
          ▼
      SQL Server

Step 3 — SQL Server Federation

For SQL Server, you create a connection to the SQL Server database. The SQL syntax follows the Unity Catalog federation model.

CREATE CONNECTION sqlserver_prod
TYPE SQLSERVER
OPTIONS (
    host '10.100.5.3',
    port '1433',
    user 'william',
    password '123456',
    database 'Customer_master_DB'
);

Step 4 — Create the Foreign Catalog

Once the connection exists:

CREATE FOREIGN CATALOG sqlserver_prod_catalog
USING CONNECTION sqlserver_prod;

SHOW CATALOGS;   

might show:
catalog_ut_internal
sqlserver_catalog
system

SHOW SCHEMAS IN sqlserver_prod_catalog;

might see
dbo
...

SHOW TABLES IN sqlserver_prod_catalog.dbo;

might see
Customer
CustomerAddress
CustomerType

You didn’t copy those tables into Delta. Databricks is exposing the external database through Unity Catalog.

Step 5 — Query the federated table

SELECT *
FROM sqlserver_prod_internal.dbo.Customer
LIMIT 10;

Step 6 — MySQL

# dummay 

# create CONNECTION 
CREATE CONNECTION mysql_prod
TYPE MYSQL
OPTIONS (
    host 'mysql-demo.company.com',
    port '3306',
    user 'dbx_federation',
    password 'YourStrongPasswordHere'
);




# create FOREIGN CATALOG
CREATE FOREIGN CATALOG mysql_sales
USING CONNECTION mysql_prod
OPTIONS (
    database 'salesdb'
);




# query 
SELECT *
FROM mysql_sales.salesdb.customers;


# different Foreign catalog
SELECT *
FROM mysql_sales.salesdb.orders o
JOIN sqlserver_prod_catalog.dbo.Customer sc
    ON o.customer_id = sc.customer_id;

Step 7 — Oracle

#CREATE CONNECTION
CREATE CONNECTION oracle_prod
TYPE ORACLE
OPTIONS (
    host 'oracle-demo.company.com',
    port '1521',
    user 'federation_user',
    password 'YourStrongPasswordHere',
    service_name 'ORCL'
);

#Oracle Foreign Catalog
CREATE FOREIGN CATALOG oracle_sales
USING CONNECTION oracle_prod;




#query
SELECT
    c.customer_id,
    c.customer_name,

    SUM(m.amount) AS mysql_sales,

    SUM(o.amount) AS oracle_sales

FROM sqlserver_federated.dbo.customers c

LEFT JOIN mysql_sales.salesdb.orders m
    ON c.customer_id = m.customer_id

LEFT JOIN oracle_sales.federation_user.orders o
    ON c.customer_id = o.customer_id

GROUP BY
    c.customer_id,
    c.customer_name
ORDER BY
    c.customer_id;

Step 8 — Azure SQL Database

# Create connection
CREATE CONNECTION azure_sql_prod
TYPE sqlserver
OPTIONS (
    host 'myazuresqlserver.database.windows.net',
    port '1433',
    user 'dbx_federation',
    password secret('my-secret-scope', 'azure-sql-password')   -- password 'YourPassword'
);


# pay attention
I used --> secret('my-secret-scope', 'azure-sql-password')

# Create catalog
CREATE FOREIGN CATALOG azure_sql_sales
USING CONNECTION azure_sql_prod
OPTIONS (
    database 'federation_lab'
);

# Query Azure SQL
SELECT *
FROM azure_sql_sales.dbo.customers;

ADLS Gen2 + Delta

Does not “connection” , “Foreign Catalog” !

Create Storage Credential

Create External Location

# Create external location 
CREATE EXTERNAL LOCATION adls_federation_location
URL 'abfss://data@mystoragelab.dfs.core.windows.net/federation/'
WITH (STORAGE CREDENTIAL adls_federation_cred);


# Create catallog



# query
SELECT *
FROM catalog_ADLS_Internal.federation_lab.sales_demo;

Blob

Does not “connection” , “Foreign Catalog” !

Create Blob External Location

# create EXTERNAL LOCATION
CREATE EXTERNAL LOCATION blob_federation_location
URL 'abfss://files@mybloblab.dfs.core.windows.net/federation/'
WITH (STORAGE CREDENTIAL blob_storage_cred);



# Create External Table for CSV
CREATE TABLE catalog_ADLS_Internal.federation_lab.marketing_campaigns
USING CSV
OPTIONS (
    header = 'true',
    inferSchema = 'true'
)
LOCATION
'abfss://files@mybloblab.dfs.core.windows.net/federation/marketing/';

# query
SELECT *
FROM catalog_ADLS_Internal.federation_lab.marketing_campaigns;

Remember this:

Source	UC Object	关键代码
SQL Server	Foreign Catalog	`CREATE CONNECTION`
MySQL	Foreign Catalog	`CREATE CONNECTION`
Oracle	Foreign Catalog	`CREATE CONNECTION`
Azure SQL DB	Foreign Catalog	`CREATE CONNECTION`
ADLS Gen2 + Delta	External Table	`CREATE EXTERNAL LOCATION`
Azure Blob	External Table	`CREATE EXTERNAL LOCATION`

Comparison of Fabric, Azure Databricks and Synapse Analytics

Microsoft Fabric vs Databricks vs Synapse

Microsoft Fabric is an all-in-one SaaS analytics platform with integrated BI.
Databricks is a Spark-based platform mainly used for large-scale data engineering and machine learning.
Synapse is an enterprise analytics service combining SQL data warehousing and big data processing.

Platform	Description (English)
Microsoft Fabric	An all-in-one SaaS data platform that integrates data engineering, data science, warehousing, real-time analytics, and BI.
Azure Databricks	A Spark-based analytics and AI platform optimized for large-scale data engineering and machine learning.
Azure Synapse Analytics	An analytics service combining data warehousing and big data analytics.

Architecture

Microsoft Fabric: Fully integrated SaaS platform built around OneLake.
single data lake, unified workspace, built-in Power BI
Databricks: Spark-native architecture optimized for big data processing.
Delta Lake, Spark clusters, ML workloads
Synapse: Hybrid analytics platform integrating SQL data warehouse and big data tools.

Main Use Cases

Platform	Best For
Fabric	End-to-end analytics platform
Databricks	Advanced data engineering & ML
Synapse	Enterprise data warehouse

Introducing Fabric

What is Fabic?

Microsoft Fabric is an all-in-one, SaaS (Software as a Service) analytics platform. It combines data movement, data engineering, data science, and business intelligence into one single website. It is built on OneLake, which is like “OneDrive for data.”

Why use Microsoft Fabric?

Traditional data stacks are “fragmented”—you might use Azure Data Factory for moving data, Databricks for cleaning it, Datalake for storing it, and Power BI for seeing it.

Fabric fixes this by putting everything in one place.

Unified Data: Every tool (SQL, Spark, Power BI) uses the same copy of data in OneLake. No more duplicating data.
SaaS Simplicity: You don’t need to manage servers, clusters, or storage accounts. It’s all managed by Microsoft.
Direct Lake Mode: Power BI can read data directly from the lake without “importing” it, making reports incredibly fast.
Cost Efficiency: You pay for one pool of “Compute Capacity” and share it across all your teams.

What can Microsoft Fabric do?

It handles the entire data journey:

Ingest: Pull data from anywhere (SQL, AWS, Web).
Store: Store massive amounts of data in an open format (Delta Parquet).
Process: Clean and transform data using Python/Spark or SQL.
Analyze: Run complex queries and train Machine Learning models.
Visualize: Build real-time dashboards in Power BI.

Components of Fabric

Fabric is divided into “Experiences” based on what you need to do:

Component (Experience)	Role	Purpose
Data Factory	Data Engineer	ETL, Pipelines, and Dataflows.
Synapse Data Engineering	Spark Developer	High-scale processing using Notebooks.
Synapse Data Warehouse	SQL Developer	Professional-grade SQL storage.
Synapse Data Science	Data Scientist	Building AI and ML models.
Real-Time Intelligence	IoT / App Dev	High-speed streaming data.
Power BI	Business Analyst	Visualizing data for the business.
Data Activator	Operations	Automatic alerts (e.g., “Email me if sales drop”).

Step-by-Step: Getting Started

If you were a Data Engineer working on a project today, your workflow would look like this:

Step 1: Create a Workspace

Log in to Fabric. Create a Workspace.
Ensure it has a Fabric Capacity license attached.

Step 2: The Lakehouse (Bronze)

Create a Lakehouse called “Company_Data_Lake“.
This creates a folder in OneLake where your raw files will live.

Step 3: Data Pipeline (Ingestion

Use Data Factory to create a pipeline.
>> Data Pipeline | Source: SQL Server | Destination: Lakehouse Files | 03/2026
This pulls your raw data into the “Files” folder.

Step 4: Notebook (Silver/Gold)

Open a Notebook. Use PySpark to clean the data.
Example: Remove duplicates, fix date formats.
Save the result as a Table (Delta format).

Step 5: Power BI (Visualization)

Switch to your SQL Analytics Endpoint.
Click New Report. Power BI will use Direct Lake to show your data instantly.

Appendix

Microsoft Fabric fundamentals documentation

Logging in Azure Databricks with Python

In Azure Databricks, logging is crucial for monitoring, debugging, and auditing your notebooks, jobs, and applications. Since Databricks runs on a distributed architecture and utilizes standard Python, you can use familiar Python logging tools, along with features specific to the Databricks environment like Spark logging and MLflow tracking.

Python’s logging module provides a versatile logging system for messages of different severity levels and controls their presentation. To get started with the logging module, you need to import it to your program first, as shown below:

import logging

logging.debug("A debug message")
logging.info("An info message")
logging.warning("A warning message")
logging.error("An error message")
logging.critical("A critical message")

log levels in Python

Log levels define the severity of the event that is being logged. For example a message logged at the INFO level indicates a normal and expected event, while one that is logged at the ERROR level signifies that some unexpected error has occurred.

Each log level in Python is associated with a number (from 10 to 50) and has a corresponding module-level method in the logging module as demonstrated in the previous example. The available log levels in the logging module are listed below in increasing order of severity:

Logging Level Quick Reference

Level	Meaning	When to Use
`DEBUG` (10)	detailed debugging info	development
`INFO` (20)	normal messages	job progress
`WARNING` (30)	minor issue	non-critical issues
`ERROR` (40)	failure	recoverable errors
`CRITICAL` (50)	system failure	stop the job

It’s important to always use the most appropriate log level so that you can quickly find the information you need. For instance, logging a message at the WARNING level will help you find potential problems that need to be investigated, while logging at the ERROR or CRITICAL level helps you discover problems that need to be rectified immediately.

By default, the logging module will only produce records for events that have been logged at a severity level of WARNING and above.

Logging basic configuration

Ensure to place the call to logging.basicConfig() before any methods such as info(), warning(), and others are used. It should also be called once as it is a one-off configuration facility. If called multiple times, only the first one will have an effect.

logging.basicConfig() example

import logging
from datetime import datetime

## set date time format
run_id = datetime.now().strftime("%Y%m%d_%H%M%S")

## Output WITH writing to a file (Databricks log file)
log_file = f"/dbfs/tmp/my_pipeline/logs/run_{run_id}.log"

## start configuing
logging.basicConfig(
    filename=log_file,
    level=logging.INFO,
    format="%(asctime)s — %(levelname)s — %(message)s",
)

## creates (or retrieves) a named logger that you will use to write log messages.
logger = logging.getLogger("pipeline")
## The string "pipeline" is just a name for the logger. It can be  anything:
## "etl"
## "my_app"
## "sales_job"
## "abc123"

simple logging example

import logging
from datetime import datetime

run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = f"/dbfs/tmp/my_pipeline/logs/run_{run_id}.log"

logging.basicConfig(
    filename=log_file,
    level=logging.INFO,
    format="%(asctime)s — %(levelname)s — %(message)s",
)
logger = logging.getLogger("pipeline")

logger.info("=== Pipeline Started ===")

try:
    logger.info("Step 1: Read data")
    df = spark.read.csv(...)
    
    logger.info("Step 2: Transform")
    df2 = df.filter(...)
    
    logger.info("Step 3: Write output")
    df2.write.format("delta").save(...)

    logger.info("=== Pipeline Completed Successfully ===")

except Exception as e:
    logger.error(f"Pipeline Failed: {e}")
    raise

the log output looks this:

2025-11-21 15:05:01,112 – INFO – === Pipeline Started ===
2025-11-21 15:05:01,213 – INFO – Step 1: Read data
2025-11-21 15:05:01,315 – INFO – Step 2: Transform
2025-11-21 15:05:01,417 – INFO – Step 3: Write output
2025-11-21 15:05:01,519 – INFO – === Pipeline Completed Successfully ===

Log Rotation (Daily Files)

Log Rotation means: Your log file does NOT grow forever. Instead, it automatically creates a new log file each day (or each hour, week, etc.), and keeps only a certain number of old files.

This prevents:

huge log files
storage overflow
long-term disk growth
difficult debugging
slow I/O

It is very common in production systems (Databricks, Linux, App Servers, Databases). Without log rotation → 1 file becomes huge, With daily rotation:

my_log.log (today)
my_log.log.2025-11-24 (yesterday)
my_log.log.2025-11-23
my_log.log.2025-11-22

Python code that does Log Rotation

import logging
from logging.handlers import TimedRotatingFileHandler

handler = TimedRotatingFileHandler(
    "/dbfs/Volumes/logs/my_log.log",
    when="midnight",   # rotate every day
    interval=1,        # 1 day
    backupCount=30     # keep last 30 days
)

formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)

logger = logging.getLogger("rotating_logger")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

logger.info("Log rotation enabled")

Hourly Log Rotation
handler = TimedRotatingFileHandler(
    "/dbfs/tmp/mylogs/hourly.log",
    when="H",       # rotate every hour
    interval=1,
    backupCount=24  # keep 24 hours
)

Size-Based Log Rotation
handler = RotatingFileHandler(
    "/dbfs/tmp/mylogs/size.log",
    maxBytes=5 * 1024 * 1024,  # 5 MB
    backupCount=5              # keep 5 old files
)

Logging to Unity Catalog Volume (BEST PRACTICE)

Create a volume first (once):

CREATE VOLUME IF NOT EXISTS my_catalog.my_schema.logs;

use it in code:

from logger_setup import get_logger

logger = get_logger(
    app_name="customer_etl",
    log_path="/dbfs/Volumes/my_catalog/my_schema/logs/customer_etl"
)

logger.info("Starting ETL pipeline")

Databricks ETL Logging Template (Production-Ready)

Features

Writes logs to file
Uses daily rotation (keeps 30 days)
Logs INFO, ERROR, stack traces
Works in notebooks + Jobs
Fully reusable

1. Create the logger (ready for copy & paste)

“File Version”

 import logging
from logging.handlers import TimedRotatingFileHandler

def get_logger(name="etl"):
    log_path = "/dbfs/tmp/logs/pipeline.log"   # or a UC Volume

    handler = TimedRotatingFileHandler(
        log_path,
        when="midnight",
        interval=1,
        backupCount=30
    )

    formatter = logging.Formatter(
        "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
    )
    handler.setFormatter(formatter)

    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)

    # Prevent duplicate handlers in notebook re-runs
    if not logger.handlers:
        logger.addHandler(handler)

    return logger

logger = get_logger("my_pipeline")
logger.info("Logger initialized")

Unity Catalog Volume version

# If the folder doesn't exist:
dbutils.fs.mkdirs("/Volumes/my_catalog/my_schema/logs")

/Volumes/my_catalog/my_schema/logs/

# Create logger pointing to the UC Volume
import logging
from logging.handlers import TimedRotatingFileHandler

def get_logger(name="etl"):
    log_path = "/Volumes/my_catalog/my_schema/logs/pipeline.log"

    handler = TimedRotatingFileHandler(
        filename=log_path,
        when="midnight",
        interval=1,
        backupCount=30           # keep last 30 days
    )

    formatter = logging.Formatter(
        "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
    )
    handler.setFormatter(formatter)

    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)

    # Prevent duplicate handlers when re-running notebook cells
    if not logger.handlers:
        logger.addHandler(handler)

    return logger

logger = get_logger("my_pipeline")
logger.info("Logger initialized")

2. Use the logger inside your ETL

logger.info("=== ETL START ===")

try:
    logger.info("Step 1: Read data")
    df = spark.read.csv("/mnt/raw/data.csv")

    logger.info("Step 2: Transform")
    df2 = df.filter("value > 0")

    logger.info("Step 3: Write output")
    df2.write.format("delta").mode("overwrite").save("/mnt/curated/data")

    logger.info("=== ETL COMPLETED ===")

except Exception as e:
    logger.error(f"ETL FAILED: {e}", exc_info=True)
    raise

Resulting log file (example)

The output file looks like this:

2025-11-21 15:12:01,233 - INFO - my_pipeline - === ETL START ===
2025-11-21 15:12:01,415 - INFO - my_pipeline - Step 1: Read data
2025-11-21 15:12:01,512 - INFO - my_pipeline - Step 2: Transform
2025-11-21 15:12:01,660 - INFO - my_pipeline - Step 3: Write output
2025-11-21 15:12:01,780 - INFO - my_pipeline - === ETL COMPLETED ===

If error:

2025-11-21 15:15:44,812 - ERROR - my_pipeline - ETL FAILED: File not found
Traceback (most recent call last):
  ...

Best Practices for Logging in Python

By utilizing logs, developers can easily monitor, debug, and identify patterns that can inform product decisions, ensure that the logs generated are informative, actionable, and scalable.

Avoid the root logger
it is recommended to create a logger for each module or component in an application.
Centralize your logging configuration
Python module that will contain all the logging configuration code.
Use correct log levels
Write meaningful log messages
% vs f-strings for string formatting in logs
Logging using a structured format (JSON)
Include timestamps and ensure consistent formatting
Keep sensitive information out of logs
Rotate your log files
Centralize your logs in one place

Conclusion

To achieve the best logging practices, it is important to use appropriate log levels and message formats, and implement proper error handling and exception logging. Additionally, you should consider implementing log rotation and retention policies to ensure that your logs are properly managed and archived.

Technical Interview Questions and Answers

The Job Interview is the single most critical step in the job hunting process. It is the definitive arena where all of your abilities must integrate. The interview itself is not a skill you possess; it is the moment you deploy your Integrated Skill Set, blending:

Hard Skills (Technical Mastery): Demonstrating not just knowledge of advanced topics, but the depth of your expertise and how you previously applied it to solve complex, real-world problems.
Soft Skills (Communication & Presence): Clearly articulating strategy, managing complexity under pressure, and exhibiting the leadership presence expected of senior-level and expert-level candidates.
Contextual Skills (Business Acumen): Framing your solutions within the company’s business goals and culture, showing that you understand the strategic impact of your work.

This Integrated Skill represents your first real opportunity to sell your strategic value to the employer.

Landing Zone

A Microsoft Landing Zone usually refers to an Azure Landing Zone (often part of the Cloud Adoption Framework (CAF)). It is a pre-built, best-practice cloud foundation that organizations use to deploy workloads in Azure in a secure, governed, and scalable way.

A Landing Zone is the “foundation environment” in the cloud:

networking is ready
security is enforced
identity is centralized
governance is in place
logging & monitoring are enabled

Core Components

Identity & Access (Entra ID)
- Central authentication
- RBAC roles
Networking
- Hub-Spoke architecture
- VNet design
- Private endpoints
Governance
- Policies (Azure Policy)
- Management Groups
- Resource tagging standards
Security
- Microsoft Defender for Cloud
- Security baselines
- Key Vault for secrets
Monitoring
- Log Analytics
- Azure Monitor
- Alerting system

Star schema vs Snowflake schema

Star schema has a central fact table connected directly to denormalized dimension tables. It is simple and fast for querying.
Snowflake schema has normalized dimension tables, where dimensions are split into multiple related tables.

Snowflake = more normalized, less redundancy but more joins

Star = simpler, faster queries

How do you design a dimensional model?

I design it using business requirements:

Align with BI/reporting needs
Identify facts (measures) like sales, transactions
Identify dimensions like customer, product, time
Choose grain (very important)
Use surrogate keys for dimensions
Apply star schema for analytics performance

Go to Top

Back to Architect

What is Azure Data Factory?

Azure Data Factory is a cloud-based data integration service used to create data-driven workflows for orchestrating and automating data movement and transformation across different data stores and compute services.

Key capabilities

Data ingestion
Data orchestration
Data transformation
Scheduling and monitoring

What are the core components of ADF?

The main components are:

Component	Purpose
Pipeline	Logical grouping of activities
Activity	Single task in a pipeline
Dataset	Data structure pointing to data
Linked Service	Connection to external resources
Trigger	Schedule or event that starts a pipeline
Integration Runtime	Compute infrastructure used to run activities

What is a Pipeline?

A pipeline is a logical grouping of activities that together perform a task such as data ingestion or transformation.

What is Integration Runtime (IR)?

Integration Runtime is the compute infrastructure used by ADF to perform data integration tasks.

Azure IR Fully managed compute in Azure
Self-hosted IR Runs on on-premises machines
Azure SSIS IR Used to run SSIS packages

Types:

What is a Linked Service?

A linked service defines the connection information needed for ADF to connect to external resources, example:

Azure SQL Database
Data Lake
databricks
On-Prem database

What is a Dataset?

A dataset represents the structure of data within a data store.

Difference between Pipeline and Data Flow

Feature	Pipeline	Data Flow
Purpose	Orchestration	Data transformation
Compute	Orchestration engine	Spark cluster
UI	Activity workflow	Visual transformation

How do you handle incremental loading?

Solution 1: Watermark column
last_modified_date:
Max(Last_modified)
WHERE last_modified > last_run_time

Solution 2: CDC
transaction log
change tracking

Solution 3: File-based incremental
folder partition by date

How do you implement error handling?

Try-Catch pattern
Failure path
Retry policy

Activity
├ success → next step
└ failure → error handling pipeline

What are triggers in ADF?

Triggers are used to automatically start pipelines.

Trigger	Purpose
Schedule trigger	Time-based execution
Tumbling window	Time-based incremental
Event trigger	Storage events

What is Tumbling Window Trigger?

A tumbling window trigger runs pipelines at fixed time intervals and processes data in discrete time windows.

How do you parameterize pipelines?

ADF supports parameters at:

Linked Service
plpeline
dataset

What are common ADF performance optimizations?

Parallel copy: pipeline success; pipeline duration; failure rate
Staging: use blob staging
Partitioning: split large tables

How do you monitor pipelines?

Monitoring options:
ADF Monitor UI
Azure Monitor
Log Analytics

Difference between Mapping Data Flow vs Pipeline

Azure Data Factory Pipeline is an orchestration layer used to coordinate activities (copy, Databricks, API calls).
Mapping Data Flow is a visual ETL engine running on Spark for transformations without coding

How do you handle retries and failure recovery?

I handle failures using:

Delta Lake ACID for safe writes and rollback support
ADF retry policies (activity-level retries)
Idempotent pipeline design (safe re-runs)
Checkpointing in streaming jobs
Logging + monitoring (Log Analytics / Databricks logs)

How do you orchestrate Databricks notebooks from ADF?

I use ADF Databricks Notebook activity:

Prefer job clusters for cost optimization
Pass parameters from ADF to notebook
Use cluster or job cluster
Chain multiple notebooks in pipeline
Use retries + monitoring in ADF

ADF retry policies (activity-level retries)

ADF supports built-in retry at the activity level:

You can configure retry count + interval
Used for transient failures (network, timeout, throttling)
Works per activity (Copy, Databricks, Web, etc.)
Combined with timeout settings for better resilience

But retries should not replace good design (idempotency is still required).

ADF → Databricks Notebook failure: how do you know what happened?

From ADF side, I rely on:

Activity output logs
- ADF captures Databricks job run URL + run ID
Error message propagation
- ADF shows Databricks notebook failure message (exit code / exception summary)
Databricks Job Run API / UI
- Use run ID to open detailed error logs in Databricks
Log Analytics integration
- Centralized monitoring if enabled
Notebook logging best practice
- Explicit try/except + logging errors back to ADF (custom messages)

So ADF gives high-level error, Databricks gives detailed stack trace.

How do you manage “Schema Drift” when ingesting data from diverse, changing corporate sources using Azure Data Factory (ADF)?

Answer: * I design a resilient ingest architecture rather than forcing hardcoded mappings. In ADF, I use Mapping Data Flows with the “Allow Schema Drift” option enabled so the pipeline doesn’t break when a source team adds a column.

When landing data into the Databricks Bronze layer, I utilize Databricks Auto Loader (cloudFiles) with Schema Evolution enabled (.option("mergeSchema", "true")). This allows Delta Lake to automatically infer and append new columns to the table schema on-the-fly, while isolating structural alerts into a monitoring log.

Go to Top

Back to Data Factory

Describe the data storage options available in Databricks.

Databricks offers several ways to store data. First, there’s the Databricks File System for storing and managing files. Then, there’s Delta Lake, an open-source storage layer that adds ACID transactions to Apache Spark, making it more reliable. Databricks also integrates with cloud storage services like AWS S3, Azure Blob Storage, and Google Cloud Storage. Plus, you can connect to a range of external databases, both relational and NoSQL, using JDBC.

What is Databricks Delta (Delta Lakehouse) and how does it enhance the capabilities of Azure Databricks?

Databricks Delta, now known as Delta Lake, is an open-source storage layer that brings ACID transactions to Apache Spark and big data workloads. It enhances Azure Databricks by providing features like:

ACID transactions for data reliability and consistency.
Scalable metadata handling for large tables.
Time travel for data versioning and historical data analysis.
Schema enforcement and evolution.
Improved performance with data skipping and Z-ordering

Are there any alternative solution that is similar to Delta lakehouse?

there are several alternative technologies that provide Delta Lake–style Lakehouse capabilities (ACID + schema enforcement + time travel + scalable storage + SQL engine). such as,

Apache Iceberg
Apache Hudi
Snowflake (Iceberg Tables / Unistore)
BigQuery + BigLake
AWS Redshift + Lake Formation + Apache Iceberg
Microsoft Fabric (OneLake + Delta/DQ/DLTS)

What is Delta Live Table? Have you used DLT?

Yes. Delta Live Tables, or DLT, is a Databricks framework for building and managing reliable data pipelines. In current Databricks terminology, it is part of Lakeflow Declarative Pipelines, but DLT is still commonly used in projects and interviews.

I have used DLT in my previous projects, mainly for production data pipelines using the Bronze, Silver, and Gold architecture.

For example, in the Bronze layer, we ingested raw data from sources such as APIs, files, and enterprise systems. In Silver, we used DLT for data cleansing, validation, deduplication, CDC processing, and business transformations. The Gold layer provided curated datasets for reporting and downstream applications.

One of the key features I used was data quality expectations. I could define rules such as required business keys, valid values, and data completeness. Depending on the requirement, invalid records could be quarantined or handled separately instead of simply being dropped, while the pipeline continued processing valid data.

DLT is also declarative, which means I define what the target datasets should look like and how they depend on upstream data, rather than manually managing every execution step. Databricks manages things such as pipeline dependencies, incremental processing, monitoring, and lineage.

I also used DLT with incremental processing and CDC patterns, including identifying new or changed records using business keys and timestamps and applying changes to Delta tables.

So, in my experience, DLT is not just a transformation tool. I used it as part of a production-grade data engineering solution, covering ingestion, transformation, data quality, incremental processing, CDC, dependency management, monitoring, and the overall Bronze-Silver-Gold architecture.

Why DLT instead of regular Databricks jobs?

I would use DLT when I need a more managed and reliable data pipeline, especially for multi-stage production pipelines.

The main advantage is that DLT provides a declarative approach. I define the data transformations and dependencies, and Databricks manages much of the pipeline execution.

It also provides built-in capabilities for data quality expectations, dependency management, incremental processing, lineage, monitoring, and pipeline recovery.

For example, if I have a Bronze-to-Silver-to-Gold pipeline with multiple dependent datasets, DLT can manage those dependencies and process the data incrementally instead of requiring me to build all of that orchestration logic myself.

Regular Databricks Jobs are still very useful, especially for custom Python or PySpark processing, batch jobs, orchestration, and tasks that don’t naturally fit a declarative pipeline.

So I would say the key difference is: Databricks Jobs are primarily an execution and orchestration mechanism, while DLT provides a more complete framework for building and operating reliable data pipelines.

Delat Live table vs Telta Table side by side

Delta Table vs Delta Live Tables (Side-by-Side)

Feature	Delta Table	Delta Live Tables (DLT)
Definition	Storage layer built on Delta Lake format	Managed ETL framework built on top of Delta Lake
Main Purpose	Store reliable data with ACID properties	Build and manage data pipelines automatically
Type	Data storage / table format	Data pipeline orchestration framework
Responsibility	User manages ETL logic manually	System manages ETL pipeline automatically
Pipeline Management	Not included	Built-in (dependency handling, orchestration)
Data Processing	Manual (SQL / PySpark / Spark jobs)	Declarative (define tables, system executes pipeline)
ETL Control	Fully controlled by developer	Partially automated (framework-managed execution)
Data Quality Rules	Not built-in	Built-in (`expect`, `expect_or_drop`, etc.)
Monitoring	External tools required	Built-in monitoring & observability
Lineage Tracking	Manual or external tools	Automatic lineage tracking
Streaming Support	Manual setup required	Native streaming support built-in
Incremental Processing	Manually implemented	Automatically handled
Error Handling / Retry	Developer must implement	Built-in retry & recovery
CI/CD Integration	Possible but manual	Fully integrated with pipelines
Underlying Storage	Parquet-based Delta format	Uses Delta Tables underneath
Use Case	Store data, build tables, SQL analytics	Build production ETL pipelines (Bronze/Silver/Gold)
Complexity Level	Lower concept complexity	Higher abstraction, easier operations

What is schema evolution in Delta Lake?

Schema evolution allows Delta tables to automatically or manually adapt when new columns are added or schema changes occur, without breaking pipelines.

What is Schema Enforcement?

Schema Enforcement means the system validates incoming data against a predefined schema before writing the data. It checks column names, data types, and structure to ensure data consistency and quality. And the system will reject the write or throw an error if incoming data does not match the expected schema.

What is Schema Contract?

A schema contract is an agreement between a data producer and a data consumer that defines the expected structure and rules of the data.

For example, it can define the required columns, data types, nullable rules, and sometimes business constraints.

The purpose is to make sure the producer and consumer have a clear and stable understanding of what the data should look like.

For example, if the contract says customer_id must be a non-null STRING, the producer should always provide the data in that format. If the producer wants to make a breaking change, such as changing the data type, that change needs to be agreed upon and managed.

So, in simple terms, a schema contract is the agreed data interface between the producer and the consumer.

What is Time Travel?

Time Travel is a feature that allows you to access and query previous versions of data in a table.

In Delta Lake, every data change (insert, update, delete) is versioned. This means the system keeps a history of changes, and you can query the table as it existed at a specific point in time or version.

SELECT * FROM table VERSION AS OF 5;
SELECT * FROM table TIMESTAMP AS OF '2024-01-01';

Explain how you can use Databricks to implement a Medallion Architecture (Bronze, Silver, Gold).

Bronze Layer (Raw Data): Ingest raw data from various sources into the Bronze layer. This data is stored as-is, without any transformation.
Silver Layer (Cleaned Data, as known Enriched layer): Clean and enrich the data from the Bronze layer. Apply transformations, data cleansing, and filtering to create more refined datasets.
Gold Layer (Aggregated Data, as known Curated layer): Aggregate and further transform the data from the Silver layer to create high-level business tables or machine learning features. This layer is used for analytics and reporting.

What Is Z-Order (Databricks / Delta Lake)?

Z-Ordering in Databricks (specifically for Delta Lake tables) is an optimization technique designed to co-locate related information into the same set of data files on disk.

OPTIMIZE mytable
ZORDER BY (col1, col2);

What Is Liquid Clustering (Databricks)?

Liquid Clustering is Databricks’ next-generation data layout optimization for Delta Lake.
It replaces (and is far superior to) Z-Order.

## At creation time:
CREATE TABLE sales
CLUSTER BY (customer_id, event_date)
AS SELECT * FROM source;

## For existing tables:
ALTER TABLE sales
CLUSTER BY (customer_id, event_date);

## Trigger the actual clustering:
OPTIMIZE sales;

## Remove clustering:
ALTER TABLE sales
CLUSTER BY NONE;

What is a Dataframe, RDD, Dataset in Azure Databricks?

Dataframe refers to a specified form of tables employed to store the data within Databricks during runtime. In this data structure, the data will be arranged into two-dimensional rows and columns to achieve better accessibility.

RDD, Resilient Distributed Dataset, is a fault-tolerant, immutable collection of elements partitioned across the nodes of a cluster. RDDs are the basic building blocks that power all of Spark’s computations.

Dataset is an extension of the DataFrame API that provides compile-time type safety and object-oriented programming benefits.

What is catching and its types?

A cache is a temporary storage that holds frequently accessed data, aiming to reduce latency and enhance speed. Caching involves the process of storing data in cache memory.

What is Spark Cache / Persist (Memory Cache)

This is the standard Apache Spark feature. It stores data in the JVM Heap (Memory).

What is Databricks Disk Cache (Local Disk)

This is a Databricks-specific optimization. It automatically stores copies of remote files (Parquet/Delta) on the local NVMe SSDs of the worker nodes.

comparing .cache() and .persist()

both .cache() and .persist() are used to save intermediate results to avoid re-computing the entire lineage. The fundamental difference is that .cache() is a specific, pre-configured version of .persist().

.cache(): This is a shorthand for .persist(StorageLevel.MEMORY_ONLY). It tries to store your data in the JVM heap as deserialized objects.

.persist(): This is the more flexible version. It allows you to pass a StorageLevel to decide exactly how and where the data should be stored (RAM, Disk, or both).

How do you optimize Databricks?

When optimizing Databricks workloads, I focus on several layers. First, I optimize data layout using partitioning and Z-ordering on Delta tables. Second, I improve Spark performance by using broadcast joins, filtering data early, and caching intermediate results. Third, I tune cluster resources such as autoscaling and Photon engine. Finally, I run Delta maintenance commands like OPTIMIZE and VACUUM to manage small files and improve query performance.

Data Layout Optimization:
Partitioning:
CREATE TABLE sales
USING DELTA
PARTITIONED BY (date)

Z-Ordering:
OPTIMIZE sales
ZORDER BY (customer_id);

Liquid Clustering:

What is Photon Engine?

Photon is a high-performance query engine built in C++ that accelerates SQL queries and data processing workloads in Azure Databricks. It improves performance by using vectorized processing and optimized execution for modern hardware.

What is Data skew,

Data skew happens when some partitions have much more data than others, causing slow tasks and performance bottlenecks.

Data skew usually happens during:

Joins
GroupBy
Aggregations

When designing a production pipeline in Databricks, how do you diagnose and mitigate data skew or memory-related issues like Out of Memory (OOM) errors?

Answer: * Diagnosis: I look at the Spark UI. If one task takes significantly longer than others (Stage Skew) or if an executor’s memory usage spikes and drops unexpectedly, it indicates data skew. If an executor JVM dies entirely, it’s an OOM error.

I optimize partitioning using Adaptive Query Execution (AQE) or implement Liquid Clustering in Delta Lake to maintain balanced, performance-aware file sizes.

Mitigation for Skew: * If joining on a skewed key, I leverage Broadcast Joins if one table is small enough.

For larger tables, I use Skew Hints in Spark SQL (/*+ SKEW('tablename', 'columnname') */) to force Spark to salt the skewed key automatically.

Mitigation for OOM: * I review the code for anti-patterns like calling .collect() on large DataFrames.

How to fix Data Skew

Salting: split one hot key into multiple keys
for example: Add a random key to spread data across partitions.
sample data:
Partition 1: US (1M rows) ← Skew, slow
Partition 2: UK (10k rows)
Partition 3: CA (5k rows)

doing this “Salting”
df = df.withColumn("salt", rand() % 10)
result:
US_0 → 200k
US_1 → 200k
US_2 → 200k
US_3 → 200k
US_4 → 200k

then re-partition by the new “salt” column
df.repartition("salt")

AFTER (Salting)
Partition 1: US_0
Partition 2: US_1
Partition 3: US_2
Partition 4: US_3
Partition 5: US_4
Better Partitioning: Repartition data using a better key.
df.repartition(“country”)
AQE:
AQE (Adaptive Query Execution) in Spark dynamically optimizes query execution during runtime.

How to enable AQE. It can adjust join strategies, merge small partitions, and handle data skew by splitting large partitions based on real execution statistics instead of static estimates.
spark.conf.set("spark.sql.adaptive.enabled", "true")

How do you handle late-arriving data?

I handle late data using:

Reprocessing logic for backfill if needed
Event-time processing (not processing-time)
Watermarking in Structured Streaming
Window-based aggregations
Delta Lake MERGE INTO for upserts

Difference between batch vs streaming in Databricks

Batch processing processes data in chunks at scheduled intervals, while streaming processes data continuously in near real-time.

Databricks uses Structured Streaming, which treats streaming as incremental batch processing with fault tolerance.

How do you manage cluster sizing and cost optimization?

I manage cost by:

Using Photon and Delta optimizations to reduce compute time
Using autoscaling clusters
Choosing right node types (memory vs compute optimized)
Using job clusters instead of all-purpose clusters
Auto-termination for idle clusters

How do you validate data quality?

I validate data quality using:

Row count validation
Null checks
Duplicate checks
Data type validation
Referential integrity checks
Business rule validation

I usually automate these checks in pipelines and log failures.

How do you monitor pipeline failures?

I monitor failures using:

ADF Monitor
Databricks job monitoring
Log Analytics / Azure Monitor
Alerts and notifications (email/Teams)
Custom logging tables

I also track pipeline run IDs for troubleshooting.

What metrics do you track in production pipelines?

Key production metrics:

Pipeline success/failure rate
Execution duration
Throughput
Data latency
Row counts
Error counts
Cluster utilization/cost
SLA compliance

These help identify performance or reliability issues.

How do you handle schema drift?

Schema drift happens when source schema changes unexpectedly.

I handle it by:

Enabling schema evolution
Using dynamic column mapping
Metadata-driven pipelines
Auto Loader schema inference
Logging and alerting schema changes
Separating mandatory vs optional columns

How would you secure and manage secrets in Azure Databricks when connecting to external data sources?

Use Azure Key Vault to store and manage secrets securely.
Integrate Azure Key Vault with Azure Databricks using Databricks-backed or Azure-backed scopes.
Access secrets in notebooks and jobs using the dbutils.secrets API.
dbutils.secrets.get(scope=”<scope-name>”, key=”<key-name>”)
Ensure that secret access policies are strictly controlled and audited.

How do you design a scalable ingestion pipeline?

I design it using a layered approach (Medallion Architecture):

Bronze: raw ingestion (batch/streaming via ADF / Auto Loader)
Silver: cleaned and standardized data
Gold: business-ready aggregated data

For scalability:

Use Auto Loader for incremental file ingestion
Use partitioning + incremental loads
Decouple ingestion and transformation
Use Delta Lake for reliability and ACID

Scenario: You need to implement a data governance strategy in Azure Databricks. What steps would you take?

Data Classification: Classify data based on sensitivity and compliance requirements.
Access Controls: Implement role-based access control (RBAC) using Azure Active Directory.
Data Lineage: Use tools like Databricks Lineage to track data transformations and movement.
Audit Logs: Enable and monitor audit logs to track access and changes to data.
Compliance Policies: Implement Azure Policies and Azure Purview for data governance and compliance monitoring.

Scenario: You need to optimize a Spark job that has a large number of shuffle operations causing performance issues. What techniques would you use?

Repartitioning: Repartition the data to balance the workload across nodes and reduce skew.
Broadcast Joins: Use broadcast joins for small datasets to avoid shuffle operations.
Caching: Cache intermediate results to reduce the need for recomputation.
Shuffle Partitions: Increase the number of shuffle partitions to distribute the workload more evenly.
Skew Handling: Identify and handle skewed data by adding salt keys or custom partitioning strategies.

Scenario: You need to migrate an on-premises Hadoop workload to Azure Databricks. Describe your migration strategy.

Assessment: Evaluate the existing Hadoop workloads and identify components to be migrated.
Data Transfer: Use Azure Data Factory or Azure Databricks to transfer data from on-premises HDFS to ADLS.
Code Migration: Convert Hadoop jobs (e.g., MapReduce, Hive) to Spark jobs and test them in Databricks.
Optimization: Optimize the Spark jobs for performance and cost-efficiency.
Validation: Validate the migrated workloads to ensure they produce the same results as on-premises.
Deployment: Deploy the migrated workloads to production and monitor their performance.

Can you explain the difference between running a PySpark workload using a Databricks Notebook versus deploying it as modular Python code via Databricks Jobs? When would you choose one over the other?

Answer: * Databricks Notebooks are excellent for ad-hoc exploration, rapid prototyping, and interactive data analysis. They combine markdown documentation with executable code blocks, making them highly visual.

Decision Criteria: For production ETL pipelines, I advocate for modular code deployed via Databricks Jobs. It allows us to implement proper software engineering practices, such as unit testing (via pytest), static code analysis, and robust CI/CD branch policies. Notebooks are kept strictly for initial R&D or orchestrating lightweight, high-level control flows.

Modular Python Code (e.g., packaged .py files/wheels) breaks logic down into reusable, unit-testable classes and functions.

What is the difference between df.collect() and df.take(n) or df.show()

Answer: * df.collect() pulls every single row from the distributed executors across the network and dumps it into the memory of the single Driver node, converting the Spark DataFrame into a local Python list. It is highly dangerous because if the data volume exceeds the Driver’s available RAM, it triggers an immediate OOM crash.

df.take(n) and df.show(n) only retrieve the first n rows. They are safe because they cap the memory allocation on the Driver, making them ideal for debugging or lightweight data sampling.

Migrate from Hive Metastore to Unity Catalog, what would be your step-by-step strategy to handle data migration, permissions, and governance?

Step 1 — Enable Unity Catalog
Step 2 — Create Catalog and Schema
Step 3 — Configure External Location

Step 4 — Assess Hive Tables: Check:

managed vs external tables
Delta vs non-Delta
table size
dependencies
downstream jobs
permissions

Step 5 — Convert Non-Delta Tables (IMPORTANT)
CONVERT TO DELTA parquet.`abfss://container/path`;
or
CONVERT TO DELTA hive_metastore.sales.orders;

Step 6 — Migrate Tables

CREATE TABLE finance.sales.orders
DEEP CLONE hive_metastore.sales.orders;

or 

CREATE TABLE finance.sales.orders AS
SELECT * FROM hive_metastore.sales.orders;

or 
SYNC TABLE finance.sales.orders
FROM hive_metastore.sales.orders;
SYNC works mainly for: external tables and metadata sync scenarios

Step 7 — Migrate Permissions
GRANT SELECT ON TABLE finance.sales.orders TO analysts;

How do you handle credential rotation and secure workspace access using Databricks Secrets and Azure Key Vault in a heavily locked-down or regulated financial services environment?

Answer: * Hardcoding passwords or keys in code is strictly prohibited. I create an Azure Key Vault-backed Secret Scope within Databricks.

All database passwords, API keys, and service principal secrets live inside Azure Key Vault. When Databricks runs a pipeline, it fetches these secrets securely at runtime using dbutils.secrets.get(). Credential rotation is handled upstream entirely within Azure Key Vault (often via automated Azure Functions), and Databricks automatically reflects the rotated values without requiring code changes.

What is Delta Sharing？have you used it?

Yes. Delta Sharing is an open protocol developed by Databricks for securely sharing data across different organizations, platforms, and cloud environments without requiring the recipient to copy or physically move the underlying data.

I have used Delta Sharing in Databricks environments to share Delta tables and curated datasets with downstream consumers. For example, instead of creating another physical copy of the data for a consumer, we can provide controlled access to the shared data through a Delta Share.

A key advantage is that the provider can control what data is shared and who can access it. We can share specific tables or views rather than giving consumers direct access to the underlying storage.

In a typical implementation, the provider creates a share and adds the required tables or views. The recipient is then granted access and can consume the shared data using supported clients, without needing access to the provider’s underlying ADLS or cloud storage.

I also see Delta Sharing as useful for cross-organization and cross-cloud data sharing, because the consumer doesn’t necessarily need to be on the same Databricks workspace or even the same cloud platform.

So the key point is: Delta Sharing provides governed data sharing without requiring traditional data copying or giving the consumer direct access to the underlying storage.

What is Unity Catalog Volumes?

Unity Catalog Volumes provide a governed way to manage non-tabular files in cloud storage through Unity Catalog.

in short, Delta tables are for structured/tabular data; Volumes are for files, unstructured.

Unity Catalog Volumes provide a governed way to access and manage files in cloud storage that are not necessarily Delta tables.

I have used Volumes in Databricks environments for file-based data such as documents, JSON, CSV, and other files that need to be accessed by data engineering or AI workloads.

There are two main types: managed volumes and external volumes. With a managed volume, Databricks manages the underlying storage. With an external volume, the data remains in an external cloud storage location such as ADLS Gen2, while Unity Catalog provides centralized governance and access control.

One of the main benefits is that users and workloads can access files through a governed Unity Catalog path such as /Volumes/<catalog>/<schema>/<volume>/..., rather than directly accessing cloud storage credentials or storage paths.

This is particularly useful when we need to apply centralized access control, auditing, and governance to file-based data.

So, in simple terms, Unity Catalog tables are primarily for governed structured data, while Volumes provide governed access to files.

Go to Top

Back to Databricks

What are Synapse SQL Workspaces and how are they used?

Synapse SQL Workspaces are the environments within Azure Synapse Analytics where users can perform data querying and management tasks. They include:

Provisioned SQL Pools: Used for large-scale, high-performance data warehousing. Users can create and manage databases, tables, and indexes, and run complex queries.
On-Demand SQL Pools: Allow users to query data directly from Azure Data Lake without creating a dedicated data warehouse. It is ideal for interactive and exploratory queries.

What is the difference between On-Demand SQL Pool and Provisioned SQL Pool?

The primary difference between On-Demand SQL Pool and Provisioned SQL Pool lies in their usage and scalability:

On-Demand SQL Pool: Allows users to query data stored in Azure Data Lake without requiring a dedicated resource allocation. It is best for ad-hoc queries and does not incur costs when not in use. It scales automatically based on query demand.
Provisioned SQL Pool: Provides a dedicated set of resources for running data warehousing workloads. It is optimized for performance and can handle large-scale data operations. Costs are incurred based on the provisioned resources and are suitable for predictable, high-throughput workloads.

How does Azure Synapse Analytics handle data integration?

Azure Synapse Analytics handles data integration through Synapse Pipelines, which is a data integration service built on Azure Data Factory. It enables users to:

Ingest Data: Extract data from various sources, including relational databases, non-relational data stores, and cloud-based services.
Transform Data: Use data flows and data wrangling to clean and transform data.
Orchestrate Workflows: Schedule and manage data workflows, including ETL (Extract, Transform, Load) processes.
Data Integration Runtime: Utilizes Azure Integration Runtime for data movement and transformation tasks.

Can you explain the concept of “Dedicated SQL Pool” in Azure Synapse Analytics?

Dedicated SQL Pool is a provisioned, high-performance relational database.

Data Storage: Data must be ingested and stored internally in a proprietary columnar format. It follows a Schema-on-Write approach.
Architecture: Uses MPP (Massively Parallel Processing) architecture. Data is sharded into 60 distributions and processed by multiple compute nodes in parallel.
Cost: Billed by the hour based on the provisioned DWUs. You can pause it when not in use to save costs.
Best For: Stable production reporting, TB/PB scale enterprise data warehousing, and high-concurrency queries needing sub-second response.

What is Serverless SQL Pool and when would you use it?

Serverless SQL Pool is an on-demand, compute-only query engine with no internal storage.

Data Location: It does not store data. Data remains in the Data Lake (ADLS Gen2) in open formats like Parquet, CSV, or JSON.
Mechanism: Uses the OPENROWSET function to query lake files directly. It follows a Schema-on-Read approach.
Cost: Billed per query based on data scanned (approx. $5 USD per TB). Cost is $0 if no queries are run.
Best For: Rapid data discovery, building a Logical Data Warehouse, and ad-hoc data validation.

It is a managed Apache Spark 3 instance easily created and configured within Azure.

Managed Cluster: You don’t manage servers; you just select the node size and the number of nodes.
Auto-Scale & Auto-Pause: It automatically scales nodes based on workload and pauses after 5 minutes of inactivity to save costs.
Language Support: Supports PySpark (Python), Spark SQL, Scala, and .NET.

Go to Top

Back to Synapse

Can you give an example of a prompt design you used to help document a complex pipeline or generate validation test cases, including the constraints you applied?

Answer: * Effective prompt design relies on clear persona, context, and explicit constraints.

Example Prompt: “You are an expert Data Architect. Act as a technical documentation writer. Analyze the following PySpark code which performs a Type 2 Slowly Changing Dimension (SCD Type 2) merge. Write a technical markdown document explaining the data lineage, column mappings, and transformation logic. Constraints: Do not use placeholder code. Do not explain standard PySpark syntax. Format the column mappings strictly in a markdown table. Highlight the idempotency mechanics of the code.”

Using Generative AI tools responsibly to improve engineering velocity. How do you use tools like GitHub Copilot or ChatGPT to scaffold notebooks, write unit tests, or optimize slow SQL queries?

I use generative AI tools like GitHub Copilot and ChatGPT as productivity accelerators across development, testing, and optimization, but always within a controlled and validated engineering workflow.

For example, when building Databricks notebooks, I use Copilot to scaffold initial PySpark structures such as data ingestion patterns, Delta table reads/writes, and common transformation templates. This helps me quickly set up a working baseline, especially for standard ETL/ELT pipelines, while I focus on business logic, data modeling, and performance design.

For unit testing, I use ChatGPT or Copilot to generate initial PySpark or SQL test cases, such as null checks, schema validation, row count verification, and data quality rules. I then refine these tests to align with data contracts and production expectations, ensuring they are meaningful and not just syntactically correct.

For SQL optimization, I often use ChatGPT to analyze slow queries and suggest improvements like better join strategies, partition pruning, or indexing considerations. However, I always validate improvements using query execution plans, Databricks Spark UI, and actual runtime benchmarks before applying changes.

In all cases, I treat AI as an assistant rather than an authority. I ensure no sensitive or PII data is included in prompts, and all AI-generated code is reviewed, tested, and validated before production deployment. The goal is to improve engineering velocity without compromising correctness, security, or governance.

What are your strict guardrails regarding Data Governance and Security when using Generative AI tools? How do you ensure that sensitive corporate data, business logic, or PII (Personally Identifiable Information) are never leaked in a prompt?

Answer: * Data privacy is paramount, especially in financial services. My absolute rule is that no actual client data, proprietary database schemas, or PII can ever enter a commercial AI prompt.

When utilizing AI for troubleshooting or code generation, I completely anonymize and abstract the prompt. I swap out proprietary table names for generic placeholders (e.g., Table_A, Table_B), obfuscate custom column names (e.g., changing account_balance_cad to metric_1), and generate synthetic mockup rows if I need to illustrate a data structure issue.

AI can occasionally “hallucinate” code or logic. What is your process for validating AI-assisted code outputs before submitting a Pull Request (PR) for a peer review?

I never take AI-generated code at face value; it must be fully verified. My validation process includes:

Unit Testing: Verify the output against expected boundaries (checking edge cases like handling of Null values or empty DataFrames). Only after it passes automated testing do I integrate it into the codebase and open a PR.

Code Review & Linting: Check for syntax errors, deprecated Spark methods, or non-existent library parameters.

Local/Dev Execution: Run the generated code block inside a isolated scratchpad environment using a small, controlled sample dataset.

Go to Top

Back to AI/ML/LLM

Can you talk about database locker?

Database locking is the mechanism a database uses to control concurrent access to data so that transactions stay consistent, isolated, and safe.

Locking prevents:

Dirty reads
Lost updates
Write conflicts
Race conditions

Types of Locks:

1. Shared Lock (S)

Used when reading data
Multiple readers allowed
No writers allowed

2. Exclusive Lock (X)

Used when updating or inserting
No one else can read or write the locked item

3. Update Lock (U) (SQL Server specific)

Prevents deadlocks when upgrading from Shared → Exclusive
Only one Update lock allowed

4. Intention Locks (IS, IX, SIX)

Used at table or page level to signal a lower-level lock is coming.

5. Row / Page / Table Locks

Based on granularity:

Row-level: Most common, best concurrency
Page-level: Several rows together
Table-level: When scanning or modifying large portions

DB engines automatically escalate:

Row → Page → Table
when there are too many small locks.

Can you talk on Deadlock?

A deadlock happens when:

Transaction A holds Lock 1 and wants Lock 2
Transaction B holds Lock 2 and wants Lock 1

Both wait on each other → neither can move → database detects → kills one transaction (“deadlock victim”).

Deadlocks usually involve one writer + one writer, but can also involve readers depending on isolation level.

How to Troubleshoot Deadlocks?

A: In SQL Server: Enable Deadlock Graph Capture

run:
ALTER DATABASE [YourDB] SET DEADLOCK_PRIORITY NORMAL;

use:
DBCC TRACEON (1222, -1);
DBCC TRACEON (1204, -1);

B: Interpret the Deadlock Graph

You will see:

Processes (T1, T2…)
Resources (keys, pages, objects)
Types of locks (X, S, U, IX, etc.)
Which statement caused the deadlock

Look for:

Two queries touching the same index/rows in different order
A scanning query locking too many rows
Missed indexes
Query patterns that cause U → X lock upgrades

C. Identify

The exact tables/images involved
The order of locking
The hotspot row or range
Rows with heavy update/contention

This will tell you what to fix.

How to Prevent Deadlocks (Practical + Senior-Level)

Always update rows in the same order
Keep transactions short
Use appropriate indexes
Use the correct isolation level
Avoid long reads before writes

Can you discuss on database normalization and denormalization

Normalization is the process of structuring a relational database to minimize data redundancy (duplicate data) and improve data integrity.

Normal Form	Rule Summary	Problem Solved
1NF (First)	Eliminate repeating groups; ensure all column values are atomic (indivisible).	Multi-valued columns, non-unique rows.
2NF (Second)	Be in 1NF, AND all non-key attributes must depend on the entire primary key.	Partial Dependency (non-key attribute depends on part of a composite key).
3NF (Third)	Be in 2NF, AND eliminate transitive dependency (non-key attribute depends on another non-key attribute).	Redundancy due to indirect dependencies.
BCNF (Boyce-Codd)	A stricter version of 3NF; every determinant (column that determines another column) must be a candidate key.	Edge cases involving multiple candidate keys.

Denormalization is the process of intentionally introducing redundancy into a previously normalized database to improve read performance and simplify complex queries.

Adding Redundant Columns: Copying a value from one table to another (e.g., copying the CustomerName into the Orders table to avoid joining to the Customer table every time an order is viewed).
Creating Aggregate/Summary Tables: Storing pre-calculated totals, averages, or counts to avoid running expensive aggregate functions at query time (e.g., a table that stores the daily sales total).
Merging Tables: Combining two tables that are frequently joined into a single, wider table.

How do you optimize slow SQL queries?

I optimize SQL by:

Checking execution plan first
Adding proper indexes (if relational DB)
Reducing data scanned (filter early)
Avoiding unnecessary joins
Using partition pruning in Spark/Delta
Using caching or materialized views
Optimizing joins (broadcast small tables)

Markdown in a Databricks Notebook

Databricks Notebook Markdown is a special version of the Markdown language built directly into Databricks notebooks. It allows you to add richly formatted text, images, links, and even mathematical equations to your notebooks, turning them from just code scripts into interactive documents and reports.

Think of it as a way to provide context, explanation, and structure to your code cells, making your analysis reproducible and understandable by others (and your future self!).

Why is it Important?

Using Markdown cells effectively transforms your workflow:

Documentation: Explain the purpose of the analysis, the meaning of a complex transformation, or the interpretation of a result.
Structure: Create sections, headings, and tables of contents to organize long notebooks.
Clarity: Add lists, tables, and links to data sources or external references.
Communication: Share findings with non-technical stakeholders by narrating the story of your data directly alongside the code that generated it.

Key Features and Syntax with Examples

1. Headers (for Structure)

Use # to create different levels of headings.

%md
# Title (H1)
## Section 1 (H2)
### Subsection 1.1 (H3)
#### This is a H4 Header

Title (H1)

Section 1 (H2)

Subsection 1.1 (H3)

This is a H4 Header

2. Emphasis (Bold and Italic)

%md
*This text will be italic*
_This will also be italic_

**This text will be bold**
__This will also be bold__

**_You can combine them_**

This text will be italic This will also be italic

This text will be bold This will also be bold

You can combine them

3. Lists (Ordered and Unordered)

Unordered List:

%md
- Item 1
- Item 2
  - Sub-item 2.1
  - Sub-item 2.2

Item 1
Item 2
- Sub-item 2.1
- Sub-item 2.2

Ordered List:

%md
1. First item
2. Second item
   1. Sub-item 2.1
3. Third item

First item
Second item
1. Sub-item 2.1
Third item

4. Links and Images

link

%md
[Databricks Website](https://databricks.com)

Mainri Inc. webside

Image

%md
![mainri Inc. Logo](https://mainri.ca/wp-content/uploads/2024/08/Logo-15-trans.png)

5. Tables

%md
| Column 1 Header | Column 2 Header | Column 3 Header |
|-----------------|-----------------|-----------------|
| Row 1, Col 1    | Row 1, Col 2    | Row 1, Col 3    |
| Row 2, Col 1    | Row 2, Col 2    | Row 2, Col 3    |
| *Italic Cell*   | **Bold Cell**   | Normal Cell     |

Column 1 Header	Column 2 Header	Column 3 Header
Row 1, Col 1	Row 1, Col 2	Row 1, Col 3
Row 2, Col 1	Row 2, Col 2	Row 2, Col 3
Italic Cell	Bold Cell	Normal Cell

6. Code Syntax Highlighting (A Powerful Feature)

%md
```python
df = spark.read.table("samples.nyctaxi.trips")
display(df)
```

```sql
SELECT * FROM samples.nyctaxi.trips LIMIT 10;
```

```scala
val df = spark.table("samples.nyctaxi.trips")
display(df)
```

7. Mathematical Equations (LaTeX)

%md


$$
f(x) = \sum_{i=0}^{n} \frac{x^i}{i!}
$$

Summary

Feature	Purpose	Example Syntax
Headers	Create structure and sections	`## My Section`
Emphasis	Add bold/italic emphasis	`bold`, `italic`
Lists	Create bulleted or numbered lists	`- Item` or `1. Item`
Tables	Organize data in a grid	`\| Header \|`
Links/Images	Add references and visuals	`[Text](URL)`
Code Blocks	Display syntax-highlighted code	`python\ncode`
Math (LaTeX)	Render mathematical formulas	`$$E = mc^2$$`

In essence, Databricks Notebook Markdown is the narrative glue that binds your code, data, and insights together, making your notebooks powerful tools for both analysis and communication.

Comparison of Microsoft Fabric, Azure Synapse Analytics (ASA), Azure Data Factory (ADF), and Azure Databricks (ADB)

Today, data engineers have a wide array of tools and platforms at their disposal for data engineering projects. Popular choices include Microsoft Fabric, Azure Synapse Analytics (ASA), Azure Data Factory (ADF), and Azure Databricks (ADB). It’s common to wonder which one is the best fit for your specific needs.

Side by Side comparison

Here’s a concise comparison of Microsoft Fabric, Azure Synapse Analytics, Azure Data Factory (ADF), and Azure Databricks (ADB) based on their key features, use cases, and differences:

Feature	Microsoft Fabric	Azure Synapse Analytics	Azure Data Factory (ADF)	Azure Databricks (ADB)
Type	Unified SaaS analytics platform	Integrated analytics service	Cloud ETL/ELT service	Apache Spark-based analytics platform
Primary Use Case	End-to-end analytics (Data Engineering, Warehousing, BI, Real-Time)	Large-scale data warehousing & analytics	Data integration & orchestration	Big Data processing, ML, AI, advanced analytics
Data Integration	Built-in Data Factory capabilities	Synapse Pipelines (similar to ADF)	Hybrid ETL/ELT pipelines	Limited (relies on Delta Lake, ADF, or custom code)
Data Warehousing	OneLake (Delta-Parquet based)	Dedicated SQL pools (MPP)	Not applicable	Can integrate with Synapse/Delta Lake
Big Data Processing	Spark-based (Fabric Spark)	Spark pools (serverless/dedicated)	No (orchestration only)	Optimized Spark clusters (Delta Lake)
Real-Time Analytics	Yes (Real-Time Hub)	Yes (Synapse Real-Time Analytics)	No	Yes (Structured Streaming)
Business Intelligence	Power BI (deeply integrated)	Power BI integration	No	Limited (via dashboards or Power BI)
Machine Learning	Basic ML integration	ML in Spark pools	No	Full ML/DL support (MLflow, AutoML)
Pricing Model	Capacity-based (Fabric SKUs)	Pay-as-you-go (serverless) or dedicated	Activity-based	DBU-based (compute + storage)
Open Source Support	Limited (Delta-Parquet)	Limited (Spark, SQL)	No	Full (Spark, Python, R, ML frameworks)
Governance	Centralized (OneLake, Purview)	Workspace-level	Limited	Workspace-level (Unity Catalog)

Key Differences

Fabric vs Synapse: Fabric is a fully managed SaaS (simpler, less configurable), while Synapse offers more control (dedicated SQL pools, Spark clusters).
ADF vs Synapse Pipelines: Synapse Pipelines = ADF inside Synapse (same engine).
ADB vs Fabric Spark: ADB has better ML & open-source support, while Fabric Spark is simpler & integrated with Power BI.

When to Use Which

Microsoft Fabric
- Best for end-to-end analytics in a unified SaaS platform (no infrastructure management).
- Combines data engineering, warehousing, real-time, and BI in one place.
- Good for Power BI-centric organizations.
Azure Synapse Analytics
- Best for large-scale data warehousing with SQL & Spark processing.
- Hybrid of ETL (Synapse Pipelines), SQL Pools, and Spark analytics.
- More flexible than Fabric (supports open formats like Parquet, CSV).
Azure Data Factory (ADF)
- Best for orchestrating ETL/ELT workflows (no compute/storage of its own).
- Used for data movement, transformations, and scheduling.
- Often paired with Synapse or Databricks.
Azure Databricks (ADB)
- Best for advanced analytics, AI/ML, and big data processing with Spark.
- Optimized for Delta Lake (ACID transactions on data lakes).
- Preferred for data science teams needing MLflow, AutoML, etc.

Which One Should You Choose?

For a fully integrated Microsoft-centric solution → Fabric
For large-scale data warehousing + analytics → Synapse
For ETL/data movement → ADF (or Synapse Pipelines)
For advanced Spark-based analytics & ML → Databricks

Data Quality Framework (DQX)

Data quality is more critical than ever in today’s data-driven world. Organizations are generating and collecting vast amounts of data, and the ability to trust and leverage this information is paramount for success. Poor data quality can have severe negative impacts, ranging from flawed decision-making to regulatory non-compliance and significant financial losses.

Key Dimensions of Data Quality (DAMA-DMBOK or ISO 8000 Standards)

A robust DQX evaluates data across multiple dimensions:

Accuracy: Data correctly represents real-world values.
Completeness: No missing or null values where expected.
Consistency: Data is uniform across systems and over time.
Timeliness: Data is up-to-date and available when needed.
Validity: Data conforms to defined business rules (e.g., format, range).
Uniqueness: No unintended duplicates.
Integrity: Relationships between datasets are maintained.

What is Data Quality Framework (DQX)

A Data Quality Framework (DQX) is an open-source framework from Databricks Labs designed to simplify and automate data quality checks for PySpark workloads on both batch and streaming data.

DAX is a structured approach to assessing, monitoring, and improving the quality of data within an organization. It define, validate, and enforce data quality rules across your data pipelines. It ensures that data is accurate, consistent, complete, reliable, and fit for its intended use. so it can be used confidently for analytics, reporting, compliance, and decision-making.

This article will explore how the DQX framework helps improve data reliability, reduce data errors, and enforce compliance with data quality standards. We will step by step go through all steps, from setup and use DQX framework in databricks notebook with code snippets to implement data quality checks.

DQX usage in the Lakehouse Architecture

In the Lakehouse architecture, new data validation should happen during data entry into the curated layer to ensure bad data is not propagated to the subsequent layers. With DQX, you can implement Dead-Letter pattern to quarantine invalid data and re-ingest it after curation to ensure data quality constraints are met. The data quality can be monitored in real-time between layers, and the quarantine process can be automated.

Credits: https://databrickslabs.github.io/dqx/docs/motivation/

Components of a Data Quality Framework

A DQX typically includes:

A. Data Quality Assessment

Profiling: Analyze data to identify anomalies (e.g., outliers, nulls).
Metrics & KPIs: Define measurable standards (e.g., % completeness, error rates).
Benchmarking: Compare against industry standards or past performance.

B. Data Quality Rules & Standards

Define validation rules (e.g., “Email must follow RFC 5322 format”).
Implement checks at the point of entry (e.g., form validation) and during processing.

C. Governance & Roles

Assign data stewards responsible for quality.
Establish accountability (e.g., who fixes issues? Who approves changes?).

D. Monitoring & Improvement

Automated checks: Use tools like Great Expectations, Talend, or custom scripts.
Root Cause Analysis (RCA): Identify why errors occur (e.g., system glitches, human input).
Continuous Improvement: Iterative fixes (e.g., process changes, user training).

E. Tools & Technology

Data Quality Tools: Informatica DQ, IBM InfoSphere, Ataccama, or open-source (Apache Griffin).
Metadata Management: Track data lineage and quality scores.
AI/ML: Anomaly detection (e.g., identifying drift in datasets).

F. Culture & Training

Promote data literacy across teams.
Encourage reporting of data issues without blame.

Using Databricks DQX Framework in a Notebook

Step by Step Implementing DQX

Step 1: Install the DQX Library

install it using the Databricks Labs CLI:

%pip install databricks-labs-dqx

# Restart the kernel after the package is installed in the notebook:
# in a separate cell run:
dbutils.library.restartPython()

Step 2: Initialize the Environment and read input data

Set up the necessary environment for running the Databricks DQX framework, including:

Importing the key components from the Databricks DQX library.

DQProfiler: Used for profiling the input data to understand its structure and generate summary statistics.
DQGenerator: Generates data quality rules based on the profiles.
DQEngine: Executes the defined data quality checks.
WorkspaceClient: Handles communication with the Databricks workspace.

Import Libraries

from databricks.labs.dqx.profiler.profiler import DQProfiler
from databricks.labs.dqx.profiler.generator import DQGenerator
from databricks.labs.dqx.engine import DQEngine
from databricks.sdk import WorkspaceClient

Loading the input data that you want to evaluate.

# Read the input data from a Delta table
input_df = spark.read.table("catalog.schema.table")

Establishing a connection to the Databricks workspace.

# Initialize the WorkspaceClient to interact with the Databricks workspace
ws = WorkspaceClient()

# Initialize a DQProfiler instance with the workspace client
profiler = DQProfiler(ws)

Profiling for data quality.

# Profile the input DataFrame to get summary statistics and data profiles

summary_stats, profiles = profiler.profile(input_df)

Generate DQX quality rules/checks

# generate DQX quality rules/checks
generator = DQGenerator(ws)
checks = generator.generate_dq_rules(profiles)  # with default level "error"

dq_engine = DQEngine(ws)

Save checks in arbitrary workspace location

# save checks in arbitrary workspace location
dq_engine.save_checks_in_workspace_file(checks, workspace_path="/Shared/App1/checks.yml")

Generate DLT expectations

# generate DLT expectations
dlt_generator = DQDltGenerator(ws)

dlt_expectations = dlt_generator.generate_dlt_rules(profiles, language="SQL")
print(dlt_expectations)

dlt_expectations = dlt_generator.generate_dlt_rules(profiles, language="Python")
print(dlt_expectations)

dlt_expectations = dlt_generator.generate_dlt_rules(profiles, language="Python_Dict")
print(dlt_expectations)

The profiler samples 30% of the data (sample ratio = 0.3) and limits the input to 1000 records by default.

Profiling a Table

Tables can be loaded and profiled using profile_table.

from databricks.labs.dqx.profiler.profiler import DQProfiler
from databricks.sdk import WorkspaceClient

# Profile a single table directly
ws = WorkspaceClient()
profiler = DQProfiler(ws)

# Profile a specific table with custom options
summary_stats, profiles = profiler.profile_table(
    table="catalog1.schema1.table1",
    columns=["col1", "col2", "col3"],  # specify columns to profile
    options={
        "sample_fraction": 0.1,  # sample 10% of data
        "limit": 500,            # limit to 500 records
        "remove_outliers": True, # enable outlier detection
        "num_sigmas": 2.5       # use 2.5 standard deviations for outliers
    }
)

print("Summary Statistics:", summary_stats)
print("Generated Profiles:", profiles)

Profiling Multiple Tables

The profiler can discover and profile multiple tables in Unity Catalog. Tables can be passed explicitly as a list or be included/excluded using regex patterns.

from databricks.labs.dqx.profiler.profiler import DQProfiler
from databricks.sdk import WorkspaceClient

ws = WorkspaceClient()
profiler = DQProfiler(ws)

# Profile several tables by name:
results = profiler.profile_tables(
    tables=["main.data.table_001", "main.data.table_002"]
)

# Process results for each table
for summary_stats, profiles in results:
    print(f"Table statistics: {summary_stats}")
    print(f"Generated profiles: {profiles}")

# Include tables matching specific patterns
results = profiler.profile_tables(
    patterns=["$main.*", "$data.*"]
)

# Process results for each table
for summary_stats, profiles in results:
    print(f"Table statistics: {summary_stats}")
    print(f"Generated profiles: {profiles}")

# Exclude tables matching specific patterns
results = profiler.profile_tables(
    patterns=["$sys.*", ".*_tmp"],
    exclude_matched=True
)

# Process results for each table
for summary_stats, profiles in results:
    print(f"Table statistics: {summary_stats}")
    print(f"Generated profiles: {profiles}")

Profiling Options

The profiler supports extensive configuration options to customize the profiling behavior.

from databricks.labs.dqx.profiler.profiler import DQProfiler
from databricks.sdk import WorkspaceClient

# Custom profiling options
custom_options = {
# Sampling options
"sample_fraction": 0.2, # Sample 20% of the data
"sample_seed": 42, # Seed for reproducible sampling
"limit": 2000, # Limit to 2000 records after sampling

# Outlier detection options
"remove_outliers": True, # Enable outlier detection for min/max rules
"outlier_columns": ["price", "age"], # Only detect outliers in specific columns
"num_sigmas": 2.5, # Use 2.5 standard deviations for outlier detection

# Null value handling
"max_null_ratio": 0.05, # Generate is_not_null rule if <5% nulls

# String handling
"trim_strings": True, # Trim whitespace from strings before analysis
"max_empty_ratio": 0.02, # Generate is_not_null_or_empty if <2% empty strings

# Distinct value analysis
"distinct_ratio": 0.01, # Generate is_in rule if <1% distinct values
"max_in_count": 20, # Maximum items in is_in rule list

# Value rounding
"round": True, # Round min/max values for cleaner rules
}

ws = WorkspaceClient()
profiler = DQProfiler(ws)

# Apply custom options to profiling
summary_stats, profiles = profiler.profile(input_df, options=custom_options)

# Apply custom options when profiling tables
tables = [
"dqx.demo.test_table_001",
"dqx.demo.test_table_002",
"dqx.demo.test_table_003", # profiled with default options
]
table_options = {
"dqx.demo.test_table_001": {"limit": 2000},
"dqx.demo.test_table_002": {"limit": 5000},
}
summary_stats, profiles = profiler.profile_tables(tables=tables, options=table_options)

Understanding output

Assuming the sample data is:

customer_id	customer_name	customer_email	is_active	start_date	end_date
1	Alice	alice@mainri.ca	1	2025-01-24	null
2	Bob	bob_new@mainri.ca	1	2025-01-24	null
3	Charlie	invalid_email	1	2025-01-24	null
3	Charlie	invalid_email	0	2025-01-24	2025-01-24

# Initialize the WorkspaceClient to interact with the Databricks workspace
ws = WorkspaceClient()

# Initialize a DQProfiler instance with the workspace client
profiler = DQProfiler(ws)

# Read the input data from a Delta table
input_df = spark.read.table("catalog.schema.table")

# Display a sample of the input data
input_df.display()

# Profile the input DataFrame to get summary statistics and data profiles
summary_stats, profiles = profiler.profile(input_df)

Upon checking the summary and profile of my input data generated, below are the results generated by DQX

print(summary_stats)

Summary of input data on all the columns in input dataframe

# Summary of input data on all the columns in input dataframe
{
  "customer_id": {
    "count": 4,
    "mean": 2.25,
    "stddev": 0.9574271077563381,
    "min": 1,
    "25%": 1,
    "50%": 2,
    "75%": 3,
    "max": 3,
    "count_non_null": 4,
    "count_null": 0
  },
  "customer_name": {
    "count": 4,
    "mean": null,
    "stddev": null,
    "min": "Alice",
    "25%": null,
    "50%": null,
    "75%": null,
    "max": "Charlie",
    "count_non_null": 4,
    "count_null": 0
  },
  "customer_email": {
    "count": 4,
    "mean": null,
    "stddev": null,
    "min": "alice@example.com",
    "25%": null,
    "50%": null,
    "75%": null,
    "max": "charlie@example.com",
    "count_non_null": 4,
    "count_null": 0
  },
  "is_active": {
    "count": 4,
    "mean": 0.75,
    "stddev": 0.5,
    "min": 0,
    "25%": 0,
    "50%": 1,
    "75%": 1,
    "max": 1,
    "count_non_null": 4,
    "count_null": 0
  },
  "start_date": {
    "count": 4,
    "count_non_null": 4,
    "count_null": 0,
    "min": "2025-01-24",
    "max": "2025-01-24",
    "mean": "2025-01-24"
  },
  "end_date": {
    "count": 4,
    "count_non_null": 1,
    "count_null": 3,
    "min": 1737676800,
    "max": 1737676800
  }
}

print(profiles)
# Default Data profile generated based on input data
DQProfile(
  name='is_not_null',
  column='customer_id',
  description=None,
  parameters=None
),
DQProfile(
  name='min_max',
  column='customer_id',
  description='Real min/max values were used',
  parameters={
    'min': 1,
    'max': 3
  }
),
DQProfile(
  name='is_not_null',
  column='customer_name',
  description=None,
  parameters=None
),
DQProfile(
  name='is_not_null',
  column='customer_email',
  description=None,
  parameters=None
),
DQProfile(
  name='is_not_null',
  column='is_active',
  description=None,
  parameters=None
),
DQProfile(
  name='is_not_null',
  column='start_date',
  description=None,
  parameters=None
)

Step 3: Understanding checks applied at data

With the below snippet, we can understand the default checks applied at input data, which generated the data profile as mentioned in previous step.

# generate DQX quality rules/checks
generator = DQGenerator(ws)
checks = generator.generate_dq_rules(profiles)

print(checks)
# Checks applied on input data
[{
  'check': {
    'function': 'is_not_null',
    'arguments': {
      'col_name': 'customer_id'
    }
  },
  'name': 'customer_id_is_null',
  'criticality': 'error'
},
{
  'check': {
    'function': 'is_in_range',
    'arguments': {
      'col_name': 'customer_id',
      'min_limit': 1,
      'max_limit': 3
    }
  },
  'name': 'customer_id_isnt_in_range',
  'criticality': 'error'
},
{
  'check': {
    'function': 'is_not_null',
    'arguments': {
      'col_name': 'customer_name'
    }
  },
  'name': 'customer_name_is_null',
  'criticality': 'error'
},
{
  'check': {
    'function': 'is_not_null',
    'arguments': {
      'col_name': 'customer_email'
    }
  },
  'name': 'customer_email_is_null',
  'criticality': 'error'
},
{
  'check': {
    'function': 'is_not_null',
    'arguments': {
      'col_name': 'is_active'
    }
  },
  'name': 'is_active_is_null',
  'criticality': 'error'
},
{
  'check': {
    'function': 'is_not_null',
    'arguments': {
      'col_name': 'start_date'
    }
  },
  'name': 'start_date_is_null',
  'criticality': 'error'
}]

Step 4: Define custom Data Quality Expectations

In addition to the automatically generated checks, you can define your own custom rules to enforce business-specific data quality requirements. This is particularly useful when your organization has unique validation criteria that aren’t covered by the default checks. By using a configuration-driven approach (e.g., YAML), you can easily maintain and update these rules without modifying your pipeline code.

For example, you might want to enforce that:

Customer IDs must not be null or empty.
Email addresses must match a specific domain format (e.x: @example.com).

# Define custom data quality expectations.
import yaml

checks_custom = yaml.safe_load("""
- check:
    arguments:
        col_name: customer_id
    function: is_not_null_and_not_empty
    criticality: error
    name: customer_id_is_null
- check:
    arguments:
        col_name: customer_email
        regex: '^[A-Za-z0-9._%+-]+@example\.com$'
    function: regex_match
    criticality: error
    name: customer_emaild_is_not_valid""")

# Validate the custom data quality checks
status = DQEngine.validate_checks(checks_custom)

# The above variable for the custom config yaml file can also be pased from workspace file path as given below:
status = DQEngine.validate_checks("path to yaml file in workspace")

# Assert that there are no errors in the validation status
assert not status.has_errors

Step 5: Applying the custom rules and generating results

Once your custom data quality rules have been defined and validated, the next step is to apply them to your input data. The DQEngine facilitates this by splitting your dataset into two categories:

Silver Data: Records that meet all quality expectations.
Quarantined Data: Records that fail one or more quality checks.

This approach allows you to separate valid and invalid data for further inspection and remediation. The valid records can proceed to downstream processes, while the quarantined records can be analyzed to determine the cause of failure (e.g., missing values, incorrect formats).

Here’s how you can apply the rules and generate the results:

# Create a DQEngine instance with the WorkspaceClient
dq_engine = DQEngine(WorkspaceClient())

# Apply quality checks and split the DataFrame into silver and quarantine DataFrames
silver_df, quarantine_df = dq_engine.apply_checks_by_metadata_and_split(input_df_1, checks_custom)

Quarantined data – Not matching the rules

Summary

In essence, data quality is no longer just an IT concern; it’s a fundamental business imperative. In today’s complex and competitive landscape, the success of an organization hinges on its ability to leverage high-quality, trusted data for every strategic and operational decision.

A Data Quality Framework (DQX) helps organizations:

Establish clear quality standards
Implement automated checks
Track and resolve issues
Ensure trust in data

https://databrickslabs.github.io/dqx/docs/motivation

https://medium.com/@nivethanvenkat28/revolutionizing-data-quality-checks-using-databricks-dqx-f2a49d83c3c6

Unity Catalog – Table Type Comparison

In Azure Databricks Unity Catalog, you can create different types of tables depending on your storage and management needs. The main table types are including Managed Tables, External Tables, Delta Tables, Foreign Tables, Streaming Tables, Live Tables (deprecated), Feature Tables, and Hive Tables (legacy). Each table type is explained in detail, and a side-by-side comparison is provided for clarity.

Side-by-Side Comparison Table

Feature	Managed Tables	External Tables	Delta Tables	Foreign Tables	Streaming Tables	Delta Live Tables (DLT)	Feature Tables	Hive Tables (Legacy)
Storage	Databricks-managed	External storage	Managed/External	External database	Databricks-managed	Databricks-managed	Managed/External	Managed/External
Location	Internal Delta Lake	Specified external path	Internal/External Delta Lake	External metastore (Snowflake, BigQuery)	Internal Delta Lake	Internal Delta Lake	Internal/External Delta Lake	Internal/External storage
Ownership	Databricks	User	Databricks/User	External provider	Databricks	Databricks	Databricks/User	Databricks (Legacy Hive Metastore)
Deletion Impact	Deletes data & metadata	Deletes only metadata	Depends (Managed: Deletes, External: Keeps data)	Deletes only metadata reference	Deletes data & metadata	Deletes data & metadata	Deletes metadata (but not feature versions)	Similar to Managed/External
Format	Delta Lake	Parquet, CSV, JSON, Delta	Delta Lake	Snowflake, BigQuery, Redshift, etc.	Delta Lake	Delta Lake	Delta Lake	Parquet, ORC, Avro, CSV
Use Case	Full lifecycle management	Sharing with external tools	Advanced data versioning & ACID compliance	Querying external DBs	Continuous data updates	ETL Pipelines	ML feature storage	Legacy storage (pre-Unity Catalog)

Table type and describes

1. Managed Tables

Managed tables are tables where both the metadata and the data are managed by Unity Catalog. When you create a managed table, the data is stored in the default storage location associated with the catalog or schema.

Data Storage and location:

Unity Catalog manages both the metadata and the underlying data in a Databricks-managed location

The data is stored in a Unity Catalog-managed storage location. Typically in an internal Delta Lake storage, e.g., DBFS or Azure Data Lake Storage

Use Case:

Ideal for Databricks-centric workflows where you want Databricks to handle storage and metadata.

Pros & Cons:

Pros: Easy to manage, no need to worry about storage locations.

Cons: Data is tied to Databricks, making it harder to share externally.

Example:

CREATE TABLE managed_table (
    id INT,
    name STRING
);

INSERT INTO managed_table VALUES (1, 'Alice');

SELECT * FROM managed_table;

2. External Tables

External tables store metadata in Unity Catalog but keep data in an external storage location (e.g., Azure Blob Storage, ADLS, S3).

Data storage and Location:

The metadata is managed by Unity Catalog, but the actual data remains in external storage (like Azure Data Lake Storage Gen2 or an S3 bucket).

You must specify an explicit storage location, e.g., Azure Blob Storage, ADLS, S3).

Use Case:

Ideal for cross-platform data sharing or when data is managed outside Databricks.

Pros and Cons

Pros: Data is decoupled from Databricks, making it easier to share.

Cons: Requires manual management of external storage and permissions.

Preparing create external table

Before you can create an external table, you must create a storage credential that allows Unity Catalog to read from and write to the path on your cloud tenant, and an external location that references it.

Requirements

In Azure, create a service principal and grant it the Azure Blob Contributor role on your storage container.
In Azure, create a client secret for your service principal. Make a note of the client secret, the directory ID, and the application ID for the client secret.

step 1: Create a storage credential

You can create a storage credential using the Catalog Explorer or the Unity Catalog CLI. Follow these steps to create a storage credential using Catalog Explorer.

In a new browser tab, log in to Databricks.
Click Catalog.
Click Storage Credentials.
Click Create Credential.
Enter example_credential for he name of the storage credential.
Set Client Secret, Directory ID, and Application ID to the values for your service principal.
Optionally enter a comment for the storage credential.
Click Save.
Leave this browser open for the next steps.

Create an external location

An external location references a storage credential and also contains a storage path on your cloud tenant. The external location allows reading from and writing to only that path and its child directories. You can create an external location from Catalog Explorer, a SQL command, or the Unity Catalog CLI. Follow these steps to create an external location using Catalog Explorer.

Go to the browser tab where you just created a storage credential.
Click Catalog.
Click External Locations.
Click Create location.
Enter example_location for the name of the external location.
Enter the storage container path for the location allows reading from or writing to.
Set Storage Credential to example_credential to the storage credential you just created.
Optionally enter a comment for the external location.
Click Save.

-- Grant access to create tables in the external location
GRANT USE CATALOG
ON example_catalog
TO `all users`;
 
GRANT USE SCHEMA
ON example_catalog.example_schema
TO `all users`;
 
GRANT CREATE EXTERNAL TABLE
ON LOCATION example_location
TO `all users`;

-- Create an example catalog and schema to contain the new table
CREATE CATALOG IF NOT EXISTS example_catalog;
USE CATALOG example_catalog;
CREATE SCHEMA IF NOT EXISTS example_schema;
USE example_schema;

-- Create a new external Unity Catalog table from an existing table
-- Replace <bucket_path> with the storage location where the table will be created
CREATE TABLE IF NOT EXISTS trips_external
LOCATION 'abfss://<bucket_path>'
AS SELECT * from samples.nyctaxi.trips;
 
-- To use a storage credential directly, add 'WITH (CREDENTIAL <credential_name>)' to the SQL statement.

There are some useful Microsoft document to be refer:

“Create an external table in Unity Catalog“
“Configure a managed identity for Unity Catalog“
“Create a Unity Catalog metastore“
“Manage access to external cloud services using service credentials“
“Create a storage credential for connecting to Azure Data Lake Storage Gen2“
“External locations“

Example



CREATE TABLE external_table (
    id INT,
    name STRING
)
LOCATION 'abfss://container@storageaccount.dfs.core.windows.net/path/to/data';

INSERT INTO external_table VALUES (1, 'Bob');

SELECT * FROM external_table;

3. Foreign Tables

Foreign tables reference data stored in external systems (e.g., Snowflake, Redshift) without copying the data into Databricks.

Data Storage and Location

The metadata is stored in Unity Catalog, but the data resides in another metastore (e.g., an external data warehouse like Snowflake or BigQuery).

It does not point to raw files but to an external system.

Use Case:

Best for querying external databases like Snowflake, BigQuery, Redshift without moving data.

Pros and Cons

Pros: No data duplication, seamless integration with external systems.

Cond: Performance depends on the external system’s capabilities.

Example

CREATE FOREIGN TABLE foreign_table
USING com.databricks.spark.snowflake
OPTIONS (
    sfUrl 'snowflake-account-url',
    sfUser 'user',
    sfPassword 'password',
    sfDatabase 'database',
    sfSchema 'schema',
    dbtable 'table'
);

SELECT * FROM foreign_table;

4. Delta Tables

Delta tables use the Delta Lake format, providing ACID transactions, scalable metadata handling, and data versioning.

Data Storage and Location

A special type of managed or external table that uses Delta Lake format.

Can be in managed storage or external storage.

Use Case:

Ideal for reliable, versioned data pipelines.

Pros and Cons

Pros: ACID compliance, time travel, schema enforcement, efficient upserts/deletes.

Cons: Slightly more complex due to Delta Lake features.

Example

CREATE TABLE delta_table (
    id INT,
    name STRING
)
USING DELTA
LOCATION 'abfss://container@storageaccount.dfs.core.windows.net/path/to/delta';

INSERT INTO delta_table VALUES (1, 'Charlie');

SELECT * FROM delta_table;

-- Time travel example
SELECT * FROM delta_table VERSION AS OF 1;

5. Feature Tables

Feature tables are used in machine learning workflows to store and manage feature data for training and inference.

Data Storage and Location

Used for machine learning (ML) feature storage with Databricks Feature Store.

Can be managed or external.

Use Case:

Ideal for managing and sharing features across ML models and teams.

Pros and Cons:

Pros: Centralized feature management, versioning, and lineage tracking.

Example:

from databricks.feature_store import FeatureStoreClient
fs = FeatureStoreClient()
fs.create_table(
    name="feature_table",
    primary_keys=["id"],
    schema="id INT, feature1 FLOAT, feature2 FLOAT",
    description="Example feature table"
)

fs.write_table("feature_table", df, mode="overwrite")
features = fs.read_table("feature_table")

6. Streaming Tables

Streaming tables are designed for real-time data ingestion and processing using Structured Streaming.

Data Location:

Can be stored in managed or external storage.

Use Case:

Ideal for real-time data pipelines and streaming analytics.

Pros and Cons

Pros: Supports real-time data processing, integrates with Delta Lake for reliability.

Cons: Requires understanding of streaming concepts and infrastructure.

Example:

CREATE TABLE streaming_table (
    id INT,
    name STRING
)
USING DELTA;

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("StreamingExample").getOrCreate()
streaming_df = spark.readStream.format("delta").load("/path/to/delta")
streaming_df.writeStream.format("delta").outputMode("append").start("/path/to/streaming_table")

Delta Live Tables (DLT)

Delta Live Tables (DLT) is the modern replacement for Live Tables. It is a framework for building reliable, maintainable, and scalable ETL pipelines using Delta Lake. DLT automatically handles dependencies, orchestration, and error recovery.

Data storage and Location:

Data is stored in Delta Lake format, either in managed or external storage.

Use Case:

Building production-grade ETL pipelines for batch and streaming data.

DLT pipelines are defined using Python or SQL.
Tables are automatically materialized and can be queried like any other Delta table.

Pros and Cons

Declarative pipeline definition.
Automatic dependency management.
Built-in data quality checks and error handling.
Supports both batch and streaming workloads.

Cons: Requires understanding of Delta Lake and ETL concepts.

Example

import dlt

@dlt.table
def live_table():
    return spark.read.format("delta").load("/path/to/source_table")

8. Hive Tables (Legacy)

Hive tables are legacy tables that use the Apache Hive format. They are supported for backward compatibility.

Data storage Location:

Can be stored in managed or external storage.

Use Case:

Legacy systems or migration projects.

Pros and Cons

Pros: Backward compatibility with older systems.
Cons: Lacks modern features like ACID transactions and time travel.

Example

CREATE TABLE hive_table (
    id INT,
    name STRING
)
STORED AS PARQUET;

INSERT INTO hive_table VALUES (1, 'Dave');
SELECT * FROM hive_table;

Final Thoughts

Use Delta Live Tables for automated ETL pipelines.

Use Feature Tables for machine learning models.

Use Foreign Tables for querying external databases.

Avoid Hive Tables unless working with legacy systems.

Summary

Managed Tables: Fully managed by Databricks, ideal for internal workflows.
External Tables: Metadata managed by Databricks, data stored externally, ideal for cross-platform sharing.
Delta Tables: Advanced features like ACID transactions and time travel, ideal for reliable pipelines.
Foreign Tables: Query external systems without data duplication.
Streaming Tables: Designed for real-time data processing.
Feature Tables: Specialized for machine learning feature management.
Hive Tables: Legacy format, not recommended for new projects.

Each table type has its own creation syntax and usage patterns, and the choice depends on your specific use case, data storage requirements, and workflow complexity.

Please do not hesitate to contact me if you have any questions at William . chen @ mainri.ca

(remove all space from the email account 😊)