Databricks Community

shwetav1407 · 4 weeks ago

How Databricks’ AI agent transforms data engineering from manual craftsmanship into conversational pipeline development

Data engineers have long accepted a painful truth: building production-grade ETL pipelines means wrestling with hundreds of lines of orchestration code, manually encoding execution order, handling incremental processing logic, and then praying nothing breaks at 2 AM. Spark Declarative Pipelines (SDP) already simplified this dramatically by letting you declare what your data should look like rather than how to get there. Now, with Genie Code in Agent mode, you don’t even have to write those declarations yourself.

In this guide, we’ll walk through building a complete medallion architecture pipeline using Genie Code and SDP — from raw ingestion through business-ready analytics — and explore the patterns that make this approach production-worthy.

What Is SDP, and Why Should You Care?

Lakeflow Spark Declarative Pipelines (SDP) is Databricks’ framework for building batch and streaming data pipelines in SQL and Python. Unlike traditional Spark jobs where you manually define execution order, manage checkpoints, and handle retries, SDP lets you declare your transformations and handles the orchestration automatically.

The key benefits that matter for real-world pipelines:

Automatic orchestration — SDP analyzes dependencies across all your source files, builds a dataflow graph, and determines the optimal execution order with maximum parallelism. It also retries failures at the most granular level possible: first the Spark task, then the flow, then the pipeline.
Incremental processing built in — Materialized views automatically process only new data and changes. No more writing MERGE statements by hand.
Data quality as code — Expectations let you define quality constraints inline, right next to your transformations.
Unified batch and streaming — Toggle between batch and streaming processing modes with a single keyword change.

Here’s what that looks like compared to traditional approaches:

The Old Way (PySpark + Manual Orchestration)

# Hundreds of lines for a simple weekly sales pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, window
from delta.tables import DeltaTable

spark = SparkSession.builder.getOrCreate()

# Step 1: Read raw data (manually handle incremental)
raw_df = spark.read.format("delta").load("/data/raw_sales")
last_processed = spark.read.format("delta") \
    .load("/checkpoints/last_ts").collect()[0][0]
new_data = raw_df.filter(col("event_time") > last_processed)

# Step 2: Clean (manually write quality checks)
cleaned = new_data.filter(
    col("amount").isNotNull() & 
    (col("amount") > 0)
)

# Step 3: Aggregate (manually handle upserts)
weekly = cleaned.groupBy(
    window("event_time", "1 week"), "region"
).agg(sum("amount").alias("total_sales"))

# Step 4: Write (manually handle merge)
target = DeltaTable.forPath(spark, "/data/weekly_sales")
target.alias("t").merge(
    weekly.alias("s"),
    "t.window = s.window AND t.region = s.region"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

# Step 5: Update checkpoint (manually track state)
# ... plus an Airflow DAG for scheduling, retries, alerting

The SDP Way (SQL)

-- The entire pipeline in a few declarations

-- Bronze: raw ingestion with Auto Loader
CREATE OR REFRESH STREAMING TABLE bronze_sales
AS SELECT * FROM STREAM read_files(
  '/data/landing/sales/',
  format => 'json',
  schema => 'event_time TIMESTAMP, region STRING, 
             product STRING, amount DOUBLE'
);

-- Silver: cleansed with quality expectations
CREATE OR REFRESH STREAMING TABLE silver_sales (
  CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW,
  CONSTRAINT not_null_region EXPECT (region IS NOT NULL) ON VIOLATION DROP ROW
)
AS SELECT
  event_time,
  region,
  product,
  amount,
  current_timestamp() AS processed_at
FROM STREAM(bronze_sales);

-- Gold: business-ready weekly aggregation
CREATE OR REFRESH MATERIALIZED VIEW gold_weekly_sales
AS SELECT
  date_trunc('week', event_time) AS week_start,
  region,
  COUNT(*) AS transaction_count,
  SUM(amount) AS total_sales,
  AVG(amount) AS avg_transaction
FROM silver_sales
GROUP BY date_trunc('week', event_time), region;

That’s it. SDP handles incremental processing, execution order, retries, and checkpoint management. The bronze and silver tables use streaming semantics (the STREAM keyword), while the gold materialized view uses batch semantics but still only reprocesses changed data.

Enter Genie Code: Your AI Data Engineering Partner

Now here’s where it gets interesting. Genie Code in Agent mode — available inside the Lakeflow Pipelines Editor — doesn’t just help you write SDP code. It can autonomously plan, generate, run, validate, and fix entire pipelines from a single natural language prompt.

How Genie Code Agent Mode Works

When you enable Agent mode in the Genie Code panel within the Lakeflow Pipelines Editor, the agent adapts its capabilities specifically for data engineering tasks. Unlike chat mode, Agent mode can:

Plan a multi-step solution and present it for your review
Search your Unity Catalog for relevant tables, schemas, and lineage
Generate SQL or Python SDP source files in the pipeline editor
Run pipeline updates and read the output datasets
Diagnose and fix errors automatically, iterating until the pipeline succeeds
Respect your Unity Catalog permissions — it can only access data you can access

The key design principle is human-in-the-loop: Genie Code proposes plans and asks for approval before executing. You can Allow, Decline, or ask it to try a different approach.

Tutorial: Building a Medallion Pipeline with Genie Code

Let’s walk through building a real pipeline — an e-commerce analytics pipeline that ingests order data, cleans and enriches it, and produces dashboards-ready metrics.

Prerequisites

A Databricks workspace with Partner-powered AI features enabled
Access to the Lakeflow Pipelines Editor
Unity Catalog configured with a target catalog and schema

Step 1: Create Your Pipeline and Open Genie Code

Navigate to Pipelines in the sidebar and create a new pipeline. Give it a name like ecommerce_analytics and set your target catalog and schema (e.g., analytics.ecommerce).

Once in the Lakeflow Pipelines Editor, open the Genie Code panel and switch to Agent mode.

Step 2: Prompt Genie Code to Build the Pipeline

Start with a descriptive prompt that tells Genie Code what you want:

Your prompt: “Build a medallion architecture pipeline for e-commerce analytics. I have raw order data landing as JSON files in /Volumes/raw_data/orders/ with fields: order_id, customer_id, product_id, quantity, unit_price, order_timestamp, and shipping_region. Create bronze ingestion with Auto Loader, silver cleansing with quality expectations, and gold aggregations for daily revenue by region and top products.”

Genie Code will create a step-by-step plan that looks something like:

Plan:
1. Search Unity Catalog for existing related tables
2. Create bronze_orders.sql — streaming table with Auto Loader
3. Create silver_orders.sql — cleaned data with expectations
4. Create gold_daily_revenue.sql — daily revenue by region
5. Create gold_top_products.sql — top products materialized view
6. Run the pipeline and validate outputs

Review the plan, ask clarifying questions if needed, then select Allow to let Genie Code proceed.

Step 3: Watch Genie Code Generate Your Pipeline

Genie Code creates each source file in your pipeline. Here’s what the generated code typically looks like:

File: bronze_orders.sql

-- Bronze layer: raw ingestion from JSON landing zone

CREATE OR REFRESH STREAMING TABLE bronze_orders
COMMENT 'Raw e-commerce orders ingested via Auto Loader'
AS SELECT
  *,
  _metadata.file_name AS source_file,
  _metadata.file_modification_time AS file_mod_time,
  current_timestamp() AS ingestion_timestamp
FROM STREAM read_files(
  '/Volumes/raw_data/orders/',
  format => 'json',
  inferColumnTypes => 'true'
);

File: silver_orders.sql

-- Silver layer: cleansed and validated orders

CREATE OR REFRESH STREAMING TABLE silver_orders (
  CONSTRAINT valid_order_id 
    EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT valid_quantity 
    EXPECT (quantity > 0 AND quantity < 10000) ON VIOLATION DROP ROW,
  CONSTRAINT valid_price 
    EXPECT (unit_price > 0) ON VIOLATION DROP ROW,
  CONSTRAINT valid_timestamp 
    EXPECT (order_timestamp IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT valid_region 
    EXPECT (shipping_region IS NOT NULL) ON VIOLATION FAIL UPDATE
)
COMMENT 'Cleansed orders with quality expectations enforced'
AS SELECT
  order_id,
  customer_id,
  product_id,
  CAST(quantity AS INT) AS quantity,
  CAST(unit_price AS DOUBLE) AS unit_price,
  CAST(quantity AS INT) * CAST(unit_price AS DOUBLE) AS line_total,
  CAST(order_timestamp AS TIMESTAMP) AS order_timestamp,
  UPPER(TRIM(shipping_region)) AS shipping_region,
  ingestion_timestamp
FROM STREAM(bronze_orders);

File: gold_daily_revenue.sql

-- Gold layer: daily revenue metrics by region

CREATE OR REFRESH MATERIALIZED VIEW gold_daily_revenue
COMMENT 'Daily revenue aggregation by shipping region'
AS SELECT
  DATE(order_timestamp) AS order_date,
  shipping_region,
  COUNT(DISTINCT order_id) AS total_orders,
  COUNT(DISTINCT customer_id) AS unique_customers,
  SUM(line_total) AS total_revenue,
  AVG(line_total) AS avg_order_value,
  SUM(quantity) AS total_units_sold
FROM silver_orders
GROUP BY DATE(order_timestamp), shipping_region;

File: gold_top_products.sql

-- Gold layer: top products by revenue

CREATE OR REFRESH MATERIALIZED VIEW gold_top_products
COMMENT 'Product performance ranked by total revenue'
AS SELECT
  product_id,
  COUNT(DISTINCT order_id) AS times_ordered,
  SUM(quantity) AS total_units,
  SUM(line_total) AS total_revenue,
  AVG(unit_price) AS avg_price
FROM silver_orders
GROUP BY product_id;

Step 4: Genie Code Runs and Validates

After generating the files, Genie Code asks for permission to run the pipeline. Once you approve it:

Triggers a pipeline update
Monitors execution across all flows
Reads the output datasets to verify data landed correctly
Reports back with row counts, any expectation violations, and the DAG structure

If something fails — say a schema mismatch in the JSON files — Genie Code diagnoses the error, proposes a fix (like adjusting the schema inference or adding a CAST), and iterates until the pipeline succeeds.

Going Deeper: Python SDP with Genie Code

While SQL is the most common approach, SDP also supports Python for more complex transformation logic. The Python API uses decorators from the pyspark.pipelines module (imported as dp).

Here’s what a Python-based silver layer might look like when you need custom transformation logic:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, upper, trim, when, lit

@dp.table(
    name="silver_orders_enriched",
    comment="Orders enriched with derived customer segments"
)
@dp.expect("valid_order_id", "order_id IS NOT NULL", on_violation="drop")
@dp.expect("valid_amount", "line_total > 0", on_violation="drop")
def silver_orders_enriched():
    return (
        spark.readStream.table("bronze_orders")
        .withColumn("line_total", col("quantity") * col("unit_price"))
        .withColumn("shipping_region", upper(trim(col("shipping_region"))))
        .withColumn(
            "customer_segment",
            when(col("line_total") >= 500, lit("premium"))
            .when(col("line_total") >= 100, lit("standard"))
            .otherwise(lit("basic"))
        )
        .withColumn("order_date", col("order_timestamp").cast("date"))
    )

You can ask Genie Code specifically for Python implementations:

Your prompt: “Add a Python-based silver transformation that enriches orders with a customer loyalty tier based on historical order count from the customers table in analytics.core.”

Genie Code will search your Unity Catalog for the customers table, understand its schema, and generate a Python file that joins and enriches appropriately.

Handling Change Data Capture (CDC)

One of SDP’s most powerful features is AUTO CDC, which handles change data capture with full support for out-of-order events. This is where things get genuinely hard in traditional pipelines — and trivial in SDP.

SQL example for CDC with SCD Type 2:

-- Streaming table to capture raw CDC events

CREATE OR REFRESH STREAMING TABLE customers_cdc_raw
AS SELECT * FROM STREAM read_files(
  '/Volumes/raw_data/customers_cdc/',
  format => 'json'
);

-- Cleansed CDC with expectations

CREATE OR REFRESH STREAMING TABLE customers_cdc_clean (
  CONSTRAINT valid_id EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
)
AS SELECT
  customer_id,
  name,
  email,
  address,
  operation,
  operation_timestamp
FROM STREAM(customers_cdc_raw);

-- Apply CDC changes with SCD Type 2 history tracking

CREATE OR REFRESH STREAMING TABLE customers;

AUTO CDC INTO customers
FROM STREAM(customers_cdc_clean)
KEYS (customer_id)
SEQUENCE BY operation_timestamp
STORED AS SCD TYPE 2;

You can prompt Genie Code with something like:

Your prompt: “Add change data capture for customer updates from Debezium CDC events. I need SCD Type 2 to track historical changes to customer addresses.”

Genie Code understands the CDC patterns and generates the appropriate AUTO CDC declarations.

Data Quality Expectations: Your Safety Net

Expectations are SDP’s built-in data quality framework. There are three violation behaviors:


Behavior What Happens Use When
ON VIOLATION DROP ROW Invalid rows are silently dropped Tolerating messy source data
ON VIOLATION FAIL UPDATE Entire pipeline update fails Critical fields that must exist
(no action specified) Invalid rows are logged but kept Monitoring without blocking

Pro tip: Use Genie Code to add expectations iteratively. After an initial pipeline run, ask:

“Analyze the bronze_orders data and suggest quality expectations for the silver layer based on the actual data distribution.”

Genie Code can read the output datasets, profile the data, and propose expectations that make sense for your actual data — not just generic null checks.

Production Patterns and Best Practices

1. Pipeline Configuration with YAML Spec

Your pipeline project uses a YAML spec file for top-level configuration:

# pipeline.yaml
name: ecommerce_analytics
target_catalog: analytics
target_schema: ecommerce
libraries:
  - path: ./bronze_orders.sql
  - path: ./silver_orders.sql
  - path: ./gold_daily_revenue.sql
  - path: ./gold_top_products.sql
configuration:
  spark.sql.shuffle.partitions: "auto"

2. Parameterize with SET

Use SET to inject environment-specific configurations:

SET env = 'production';
SET raw_path = '/Volumes/${env}/raw_data/orders/';

CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  '${raw_path}',
  format => 'json'
);

3. Mix SQL and Python Files

A single pipeline can contain both SQL and Python source files. Use SQL for straightforward transformations and Python when you need UDFs, ML feature engineering, or complex business logic.

4. Use Genie Code for Ongoing Maintenance

Genie Code doesn’t just build pipelines — it monitors them. It can:

Triage failures when a pipeline run breaks
Investigate anomalies in data quality metrics
Handle DBR upgrades by updating deprecated syntax
Optimize resource allocation based on observed workload patterns

Ask it things like:

“The silver_orders pipeline has been failing since yesterday. Diagnose the issue.”

“Optimize the compute configuration for this pipeline — it’s running slowly on large backfills.”

Wrapping Up

The combination of SDP and Genie Code represents a genuine paradigm shift for data engineering on Databricks. SDP eliminates the boilerplate of pipeline orchestration, and Genie Code eliminates the boilerplate of writing SDP. What used to take days of manual pipeline construction can now happen in a single conversation.

The key takeaways:

Start with SDP — even without Genie Code, the declarative approach saves enormous amounts of manual orchestration code.
Use Genie Code Agent mode in the Lakeflow Pipelines Editor to plan, generate, and validate entire pipelines from natural language.
Build iteratively — start with a basic bronze-silver-gold structure, then ask Genie Code to add CDC handling, expectations, and enrichments.
Trust the loop — Genie Code’s ability to run the pipeline, read outputs, diagnose errors, and fix them autonomously is the real superpower.
Keep humans in control — every destructive action requires your approval. Genie Code proposes; you decide.

SDP and Genie Code are both generally available today at no additional cost for all Databricks customers. Open the Lakeflow Pipelines Editor, flip on Agent mode, and start talking to your data infrastructure.

Ready to get started? Check out the Databricks SDP documentation and the Genie Code guide for pipeline development.

Mycimmunity · 3 weeks ago

Great Overview, Thanks for sharing with the community.