iyashk-DB
Databricks Employee
Databricks Employee

‎12-03-2025 10:56 AM

DLT analyzes your code to build a dependency graph (DAG) and schedules independent flows concurrently up to the available compute; you don’t have to orchestrate parallelism yourself if flows don’t depend on each other.

parameterise a list of table names and generate per‑table flows (Python)

Use a pipeline configuration parameter (for example, table_list) and read it from your notebook. Then, create DLT tables in a loop using a small function factory so each table gets its own definition, which DLT will parallelize when they’re independent.

# Python (DLT)
import dlt
from pyspark.sql.functions import *

# 1) Read list of tables from pipeline parameter "table_list", e.g., "customers,orders,products"
tables = [t.strip() for t in spark.conf.get("table_list").split(",")]

# 2) Use a function factory to avoid late-binding issues in loops
def define_bronze(name: str):
    @dlt.table(name=f"{name}_bronze", comment=f"Bronze ingestion for {name}")
    def _bronze():
        # Example: Auto Loader per-table path; adapt format/path/options to your sources
        return (
            spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("inferSchema", True)
            .load(f"/mnt/data/{name}")  # e.g., one folder per table name
        )
    return _bronze

def define_silver(name: str):
    @dlt.table(name=f"{name}_silver", comment=f"Silver cleansing for {name}")
    def _silver():
        # Example transformation; replace with your logic
        return dlt.read_stream(f"{name}_bronze").select("*")
    return _silver

# 3) Instantiate a bronze+silver flow for each table name
for n in tables:
    define_bronze(n)
    define_silver(n)

Because DLT evaluates decorators lazily, you must create datasets inside separate functions when looping; otherwise, you’ll accidentally capture the last loop variable value for all tables.

0 Kudos