Skip to content

Data Ingestion

All data enters the Lakehouse through the Bronze layer — raw, unmodified copies of source data stored as Delta tables in S3.

Ingestion patterns

Batch ingestion with AWS DMS

Use AWS Database Migration Service to replicate relational databases (RDS, Aurora) into S3:

  1. Create a DMS replication instance in the same VPC as the source database.
  2. Configure a Full Load + CDC task to write Parquet files to s3://…/bronze/<source_name>/.
  3. An Autoloader job on Databricks picks up the files and appends them to the Bronze Delta table.
# Autoloader example — reads new files as they land in S3
(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.schemaLocation", "/checkpoints/<table>/schema")
    .load("s3://my-bucket/bronze/orders/")
    .writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoints/<table>/")
    .trigger(availableNow=True)
    .toTable("bronze_catalog.orders.orders_raw"))

Streaming ingestion with Kinesis

Real-time data flows through Amazon Kinesis Data Streams and is written to Bronze via a Databricks Structured Streaming job:

from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StringType

schema = StructType()  # define your schema

(spark.readStream
    .format("kinesis")
    .option("streamName", "my-events-stream")
    .option("initialPosition", "TRIM_HORIZON")
    .option("region", "us-east-1")
    .load()
    .select(from_json(col("data").cast("string"), schema).alias("payload"))
    .select("payload.*")
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/events/")
    .toTable("bronze_catalog.events.events_raw"))

File drop ingestion

For third-party files (CSV, JSON, XML) delivered via SFTP or directly to S3:

  1. Files land in s3://…/landing/<feed_name>/.
  2. An S3 Event Notification triggers a Lambda that copies files to s3://…/bronze/<feed_name>/.
  3. Autoloader picks up the files and ingests them.

Bronze layer conventions

Convention Value
Table location s3://…/bronze/<domain>/<table>/
Partition column _ingest_date (date the row was loaded)
Added metadata columns _source_file, _ingest_timestamp
Schema evolution mergeSchema = true — new columns are added automatically
Retention Raw files retained for 7 years; Delta log compacted monthly

Schema registry

All source schemas are documented in the schemas/ folder of the notebooks repository and enforced using Delta's columnMapping feature.

Monitoring ingestion jobs

  • Each ingestion job emits custom metrics to Amazon CloudWatch under the LakehouseIngestion namespace.
  • Alerts fire to the #lakehouse-alerts Slack channel when a job fails or lags more than 15 minutes.