Databricks Data Engineer Professional

Bundle project layout

A Declarative Automation Bundle is a databricks.yml at the project root plus the code it deploys: resources/ for job and pipeline YAML, src/ for the package (built as a wheel), tests/ for pytest. Pull resource files in with include, so dozens of jobs and pipelines stay reviewable.

bundle:
  name: orders

include:
  - resources/*.yml

targets:
  dev:
    mode: development
    default: true
  prod:
    mode: production

Keep secrets out of the bundle and out of Git: reference them as {{secrets/scope/key}} in a Spark config or environment variable, or read them with dbutils.secrets.get. databricks bundle validate checks the configuration without deploying.

Streaming table or materialized view

Joins in a streaming table don't recompute when a dimension changes: rows already written keep the old attribute. Changing a streaming table's query only affects new rows until a full refresh.

Test the code

from pyspark.testing import assertDataFrameEqual
from pyspark.testing import assertSchemaEqual

def clean(df):
    return df.dropna(subset=["id"])

def test_clean(spark):
    src = spark.createDataFrame(
        [(1, "a"), (None, "b")], "id INT, v STRING")
    out = src.transform(clean)
    want = spark.createDataFrame(
        [(1, "a")], "id INT, v STRING")
    assertSchemaEqual(out.schema, want.schema)
    assertDataFrameEqual(out, want)

Unit tests check one function on small, synthetic DataFrames. Integration tests run the connected stages end to end on sample data, for example a bundle deployed to a test target. Write each step as a function that takes and returns a DataFrame, then chain them with df.transform(clean).transform(enrich).

Stateful streaming

from pyspark.sql import functions as F

counts = (spark.readStream.table("main.iot.events")
    .withWatermark("event_time", "15 minutes")
    .groupBy(F.window("event_time", "10 minutes"))
    .count())

(counts.writeStream
    .outputMode("append")
    .option("checkpointLocation", chk)
    .toTable("main.iot.counts_10m"))

• The checkpoint stores source offsets and state; with an idempotent sink such as Delta it gives exactly-once results after a restart. One checkpoint per query, never shared.
• foreachBatch is at least once. For idempotent Delta writes inside it, set txnAppId and txnVersion (= batchId). MERGE from a stream needs foreachBatch.
• Stream-stream outer joins need watermarks; with several watermarks the default policy takes the min, so a lagging stream's data isn't dropped.
• dropDuplicatesWithinWatermark drops duplicates that arrive within the watermark.