Databricks Delta Live Tables (DLT) Python Stubs

Warnings

• gotcha The `databricks-dlt` PyPI library provides Python *stubs* for local development tools (IDE autocompletion, type checking) and *does not contain functional implementations*. DLT pipelines defined using this stub must be deployed and executed on a Databricks workspace; they cannot be run locally.
  Fix: Understand that `databricks-dlt` is a development-time aid. To run your pipelines, you need a Databricks workspace and the DLT runtime. Refer to Databricks documentation for deploying and running DLT pipelines.
• deprecated The underlying product "Delta Live Tables (DLT)" has been rebranded to "Lakeflow Spark Declarative Pipelines (SDP)". While existing Python code using `import dlt` will continue to function, Databricks officially recommends migrating new development to use `from pyspark import pipelines as dp` and the corresponding `@dp` decorators and functions for future compatibility and to leverage new features.
  Fix: For new DLT/SDP development, use `from pyspark import pipelines as dp` at the top of your Python pipeline files. Replace `@dlt.table` with `@dp.table`, `@dlt.streaming_table` with `@dp.create_streaming_table`, `@dlt.materialized_view` with `@dp.materialized_view`, etc.
• gotcha The `dlt` (or `pyspark.pipelines`) module and its decorators are only available when your Python code is executed within the context of a Databricks DLT/SDP pipeline. Attempting to import or use these modules in a standalone Python script or a regular Databricks notebook (not configured as a DLT pipeline) will result in an `ImportError` or `NameError`.
  Fix: Ensure your Python code is part of a DLT/SDP pipeline definition within a Databricks workspace. Local development with the `databricks-dlt` stub is for syntax validation only, not execution.
• gotcha Managing external Python dependencies within DLT/SDP pipelines directly with `%pip install` or init scripts on the cluster can be problematic due to potential conflicts and maintenance issues. This can lead to unexpected pipeline failures or inconsistent environments.
  Fix: Follow Databricks best practices for dependency management: package Python dependencies as a wheel or egg and specify them in the pipeline libraries settings, or use a `requirements.txt` file attached to the pipeline configuration. Avoid dynamic installation methods within the pipeline code.

Install

• pip install databricks-dlt Install the DLT stub library

Imports

• dlt

  import dlt

  While 'import dlt' is correct for existing DLT code, Databricks now recommends 'from pyspark import pipelines as dp' for new Lakeflow Spark Declarative Pipelines (SDP) development.
• pipelines (as dp)

  from pyspark import pipelines as dp

  This is the recommended import path for new Lakeflow Spark Declarative Pipelines (SDP) development, replacing the 'dlt' module.

Quickstart

This quickstart demonstrates a typical Delta Live Tables pipeline structure using Python decorators within a Databricks notebook. It outlines a medallion architecture (bronze, silver, gold) for data processing. Note that while the `databricks-dlt` library provides IDE support for such code, actual execution and data processing occur only when deployed and run as a DLT pipeline on a Databricks workspace.

# This code snippet is designed to run within a Databricks DLT Notebook environment.
# The 'databricks-dlt' library provides stubs for local development, 
# but actual execution requires a Databricks workspace.

import dlt
from pyspark.sql.functions import *

# Define a streaming table (Bronze layer)
@dlt.table
def raw_data():
    # In a real scenario, this would read from a source like Auto Loader
    # e.g., spark.readStream.format('cloudFiles').option('cloudFiles.format', 'json').load('/databricks-datasets/retail-org/sales_orders/')
    # For demonstration, we'll simulate a static DataFrame read as this is a stub example
    return spark.read.format('json').load('/databricks-datasets/retail-org/sales_orders/')

# Define a cleansed table (Silver layer) with expectations
@dlt.table(comment='Cleansed sales orders with valid order numbers')
@dlt.expect_or_drop('valid_order_number', 'order_number IS NOT NULL')
def cleansed_data():
    return dlt.read('raw_data').select(col('customer_id'), col('order_number'), col('order_date'))

# Define an aggregated table (Gold layer)
@dlt.table(name='daily_sales_summary')
def daily_sales():
    return (
        dlt.read('cleansed_data')
        .groupBy('order_date')
        .agg(count('order_number').alias('total_orders'))
    )