adnan_alvee
Databricks Employee
Databricks Employee

yesterday


Use one Auto Loader streaming table per object type, generated from a governed configuration registry;  not dbutils.fs.ls. Each object type needs an independent schema and checkpoint. Auto Loader cannot use one CSV stream to infer unrelated schemas and dynamically choose target tables: CSV inference produces one global schema for the stream.

Prefer changing the landing layout to:

<landing>/object_type=customer/...
<landing>/object_type=order/...
If that is impossible, keep the shared directory and assign each stream a non-overlapping pathGlobFilter.

Maintain a small registry containing:

  • Validated object type
  • Expected CSV schema
  • Target table
  • Optional CSV parsing options


Lakeflow Spark Declarative Pipelines can read this registry during pipeline planning and generate multiple streaming tables programmatically. The definitions are evaluated serially, but the resulting flows can execute in parallel. New objects become active on the next pipeline update. The registry should remain additive because removing a generated dataset causes that dataset to be dropped from the pipeline target schema.

Each generated flow:

  1. Reads the landing path with Auto Loader.
  2. Selects only its filenames using pathGlobFilter.
  3. Applies that object’s explicit schema.
  4. Writes to its own streaming table with independently managed state.


Auto Loader supports pre-compressed CSV files, including gzip. Lakeflow manages checkpoint and schema locations automatically; standalone streams require unique durable locations for each workload.

For production discovery, enable managed file events. They share one notification queue per external location, require Unity Catalog and DBR 14.3 LTS or later, and should run at least every seven days to avoid falling back to directory listing.

Minimal implementation example

import re
from pyspark import pipelines as dp

SOURCE = "/Volumes/<source_catalog>/<source_schema>/<landing_volume>"

# Small governed registry:
# object_type STRING, schema_ddl STRING, enabled BOOLEAN
objects = (
    spark.table("<config_catalog>.<config_schema>.ingestion_objects")
         .where("enabled = true")
         .select("object_type", "schema_ddl")
         .collect()
)

def define_object_table(object_type: str, schema_ddl: str):
    # Prevent target-name or glob injection.
    if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_]*", object_type):
        raise ValueError(f"Invalid object_type: {object_type}")

    @DP.table(name=f"bronze_{object_type.lower()}")
    def ingest_object():
        return (
            spark.readStream
                 .format("cloudFiles")
                 .option("cloudFiles.format", "csv")
                 .option("header", "true")
                 .option(
                     "pathGlobFilter",
                     f"{object_type}_????_??_??_*.csv.gz"
                 )
                 .option("cloudFiles.useManagedFileEvents", "true")
                 .option("rescuedDataColumn", "_rescued_data")
                 .schema(schema_ddl)
                 .load(SOURCE)
                 .selectExpr(
                     "*",
                     "_metadata.file_path AS _source_file",
                     "current_timestamp() AS _ingested_at"
                 )
        )

for row in objects:
    define_object_table(row.object_type, row.schema_ddl)


Another alternative could be to look into DLT-Meta, which is a datbaricks labs project. Its a meta data driven pipeline config for spark declarative pipelines. https://docs.databricks.com/aws/en/ldp/developer/dlt-meta

0 Kudos