Ingest Data into Databricks with COPY INTO, Auto Loader, and Lakeflow Connect

• COPY INTO
• Auto Loader
  • Schema enforcement and evolution
  • availableNow vs once
• Lakeflow Connect
• JDBC ingestion
• Choosing the right method

Databricks has four main ingestion methods. The right one depends on the source type, load frequency, and whether schema discovery is needed.

COPY INTO

COPY INTO is an idempotent SQL command for loading files from cloud storage into a Delta table. On each run it reads only new files — files already loaded are tracked and skipped.

COPY INTO catalog.bronze.raw_orders
FROM 's3://my-bucket/orders/'
FILEFORMAT = PARQUET
COPY_OPTIONS ('mergeSchema' = 'true');

COPY INTO works well for scheduled batch loads where the source is a known cloud storage location. It does not require a running stream — each call is a one-time read of whatever files have not been processed yet.

Supported formats include Parquet, CSV, JSON, Avro, ORC, and text. For CSV and JSON, format options handle headers, delimiters, and multi-line records:

COPY INTO catalog.bronze.raw_events
FROM 's3://my-bucket/events/'
FILEFORMAT = JSON
FORMAT_OPTIONS ('multiLine' = 'true');

Auto Loader

Auto Loader uses the cloudFiles format in Structured Streaming to ingest files as they arrive in cloud storage. Unlike COPY INTO, Auto Loader runs as a continuous stream and supports two discovery modes:

• Directory listing: polls the storage path at an interval; simpler to set up, works for most use cases
• File notification: uses cloud-native event notifications (S3 Event Notifications, Azure Event Grid) for lower-latency detection and better performance at high file volumes

df = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoints/raw_events/schema")
    .load("s3://my-bucket/events/")
)

(
    df.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoints/raw_events/")
    .trigger(availableNow=True)
    .toTable("catalog.bronze.raw_events")
)

Schema enforcement and evolution

Auto Loader handles schema changes in a few ways:

• Schema enforcement (default): if incoming files have columns not present in the existing schema, the new columns land in a _rescued_data column rather than failing the stream
• Schema evolution (cloudFiles.schemaEvolutionMode = 'addNewColumns'): new columns are automatically added to the target table
• Schema hints: manually specify expected column types when the source schema is unstable

.option("cloudFiles.schemaEvolutionMode", "addNewColumns")

The schemaLocation stores the inferred schema so Auto Loader does not re-infer it on every restart.

availableNow vs once

When using trigger(availableNow=True), Auto Loader processes all files available at start time and then stops. This makes it behave like a batch job rather than a long-running stream — useful for scheduled runs on a Job Cluster.

trigger(once=True) is the older equivalent and is deprecated in favor of availableNow.

Lakeflow Connect

Lakeflow Connect is Databricks’ managed connector service for external data sources. It covers three connector types:

• Standard connectors: built into the Databricks platform (Salesforce, Google Analytics, relational databases)
• Managed connectors: Fivetran-backed connectors that run within Databricks, with broader source coverage
• Partner connectors: third-party integrations available from the Databricks Marketplace

Lakeflow Connect connectors run as Lakeflow Jobs and write to Unity Catalog tables. There is no infrastructure to manage — the ingestion pipeline is configured through the UI or API and Databricks handles the scheduling and compute.

JDBC ingestion

For database sources (MySQL, PostgreSQL, SQL Server, Oracle), JDBC is the standard ingestion path:

df = (
    spark.read
    .format("jdbc")
    .option("url", "jdbc:postgresql://host:5432/mydb")
    .option("dbtable", "orders")
    .option("user", "dbuser")
    .option("password", dbutils.secrets.get("scope", "db-password"))
    .load()
)

By default, JDBC reads the entire table on a single executor — slow for large tables. Parallelism requires four options:

df = (
    spark.read
    .format("jdbc")
    .option("url", "jdbc:postgresql://host:5432/mydb")
    .option("dbtable", "orders")
    .option("partitionColumn", "order_id")
    .option("lowerBound", "1")
    .option("upperBound", "1000000")
    .option("numPartitions", "10")
    .option("user", "dbuser")
    .option("password", dbutils.secrets.get("scope", "db-password"))
    .load()
)

partitionColumn must be a numeric or date column. The range from lowerBound to upperBound is split into numPartitions chunks, each read by a separate executor. The bounds do not filter the data — rows outside the range are still included — they only determine how the work is divided.

Choosing the right method