We want to write the data from delta live tables to event hubs using managed identity

Architecture

DLT (or Delta table) → Structured Streaming → Event Hubs (AAD MI)

• DLT pipeline continuously writes a Delta table (e.g., catalog.schema.silver_orders).
• A separate streaming job/notebook reads that table (as a stream) and publishes JSON records to Event Hubs using Managed Identity via the Azure SDK.

Why separate? DLT’s native sink is Delta; using a separate streaming job keeps your DLT clean and lets you manage backpressure/retries independently.

Notebook A — Delta → Event Hubs (Managed Identity)

Save as: notebooks/streaming/DeltaToEventHubs_MI.py

# Databricks notebook source
# MAGIC %md
# MAGIC # Delta → Event Hubs via Managed Identity (Structured Streaming)
# MAGIC This notebook reads from a Delta/DLT **streaming** table and publishes JSON rows to Azure Event Hubs using AAD Managed Identity.
# MAGIC
# MAGIC ### What you need configured
# MAGIC - Databricks Workspace **Managed Identity** (system-assigned or user‑assigned) enabled.
# MAGIC - MI has **Azure Event Hubs Data Sender** role on the EH **namespace** (recommended) or specific event hub.
# MAGIC - Unity Catalog permissions to read the Delta table.
# MAGIC - Network access from cluster to Event Hubs endpoint.
# MAGIC
# MAGIC ### How auth works
# MAGIC Uses `azure-identity` → `ManagedIdentityCredential` (or `DefaultAzureCredential`) to acquire AAD tokens inside executors.

# COMMAND ----------

# MAGIC %md
# MAGIC ## Install libraries (cluster-scoped alternative is also fine)
# MAGIC For runtime reproducibility, install from the notebook. Pin versions tested with Spark 3.x.
# MAGIC - azure-identity: token acquisition for MI
# MAGIC - azure-eventhub: native producer client with batching

# COMMAND ----------

# MAGIC %pip install --quiet "azure-identity==1.17.1" "azure-eventhub==5.15.0"
# MAGIC dbutils.library.restartPython()

# COMMAND ----------

# After restart, import dependencies.
import os
import json
from typing import Iterable

from azure.identity import ManagedIdentityCredential, DefaultAzureCredential
from azure.eventhub import EventHubProducerClient, EventData

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import to_json, struct, col, lit
from pyspark.sql.streaming import StreamingQuery

spark: SparkSession = spark

# COMMAND ----------

# MAGIC %md
# MAGIC ## Parameters (use widgets for convenience)
# MAGIC Populate these for your environment. If using **user‑assigned MI**, set its client ID below (or as a secret).

# COMMAND ----------

dbutils.widgets.text("catalog",            "main",   "Catalog")
dbutils.widgets.text("schema",             "stream", "Schema")
dbutils.widgets.text("table",              "silver_orders", "Delta/ DLTable name")
dbutils.widgets.text("checkpoint_path",    "dbfs:/pipelines/eh_out/checkpoints/silver_orders", "Checkpoint path")
dbutils.widgets.text("eh_fq_namespace",    "<eh-namespace>.servicebus.windows.net", "Event Hubs FQ Namespace")
dbutils.widgets.text("eh_name",            "<event-hub-name>", "Event Hub name")
dbutils.widgets.text("user_assigned_mi_client_id", "", "User-Assigned MI Client ID (optional)")
dbutils.widgets.text("max_events_per_batch", "1000", "Max events per EH batch flush")

catalog          = dbutils.widgets.get("catalog")
schema           = dbutils.widgets.get("schema")
table            = dbutils.widgets.get("table")
checkpoint_path  = dbutils.widgets.get("checkpoint_path")
eh_fq_namespace  = dbutils.widgets.get("eh_fq_namespace")
eh_name          = dbutils.widgets.get("eh_name")
uami_client_id   = dbutils.widgets.get("user_assigned_mi_client_id").strip() or None
max_events_batch = int(dbutils.widgets.get("max_events_per_batch"))

source_table_fqn = f"{catalog}.{schema}.{table}"

display({
  "source_table": source_table_fqn,
  "checkpoint_path": checkpoint_path,
  "event_hubs_fqdn": eh_fq_namespace,
  "event_hub": eh_name,
  "uami_client_id": uami_client_id or "(system-assigned MI)"
})

# COMMAND ----------

# MAGIC %md
# MAGIC ## Managed Identity credential helper
# MAGIC On Databricks, **ManagedIdentityCredential** is the most explicit.
# MAGIC If a **User-Assigned MI** is used, pass its `client_id`.
# MAGIC Fallback to `DefaultAzureCredential` for flexibility in dev/test.

# COMMAND ----------

def build_credential(uami_client_id: str | None):
    try:
        if uami_client_id:
            return ManagedIdentityCredential(client_id=uami_client_id)
        # System-assigned MI:
        return ManagedIdentityCredential()
    except Exception as mi_err:
        # Dev fallback (e.g., local testing with Azure CLI login)
        print(f"[WARN] ManagedIdentityCredential failed: {mi_err}. Falling back to DefaultAzureCredential.")
        return DefaultAzureCredential(exclude_shared_token_cache_credential=True)

# COMMAND ----------

# MAGIC %md
# MAGIC ## Efficient partition writer
# MAGIC Uses `foreachBatch` → `foreachPartition` to avoid driver collects.
# MAGIC Creates **EventDataBatch** and flushes when full (size-aware).

# COMMAND ----------

def send_partition(iterator: Iterable[str],
                   fq_namespace: str,
                   eventhub_name: str,
                   uami_client_id: str | None):
    """Send JSON strings from a partition to Event Hubs."""
    credential = build_credential(uami_client_id)
    producer = EventHubProducerClient(
        fully_qualified_namespace=fq_namespace,
        eventhub_name=eventhub_name,
        credential=credential
    )
    try:
        batch = producer.create_batch()
        for json_str in iterator:
            try:
                # Ensure bytes/string payload
                event = EventData(json_str)
                # Add to current batch or flush and create a new one
                if not batch.try_add(event):
                    producer.send_batch(batch)
                    batch = producer.create_batch()
                    # second attempt must succeed for reasonable payload sizes
                    if not batch.try_add(event):
                        # If still too large, drop or log (rare: single row too big)
                        # Log and skip to avoid blocking the stream.
                        print(f"[WARN] Skipped oversize single event (len={len(json_str)})")
            except Exception as e:
                # Non-fatal per-record error
                print(f"[WARN] Event add error: {e}. Skipping record.")
        # Flush remainder
        if len(batch) > 0:
            producer.send_batch(batch)
    finally:
        producer.close()

def foreach_batch_writer(batch_df: DataFrame, batch_id: int):
    # Convert entire micro-batch to compact JSON strings
    json_ds = batch_df.select(to_json(struct(*[col(c) for c in batch_df.columns])).alias("json")).rdd.map(lambda r: r["json"])
    # Send partition-wise
    json_ds.foreachPartition(lambda it: send_partition(it, eh_fq_namespace, eh_name, uami_client_id))

# COMMAND ----------

# MAGIC %md
# MAGIC ## Build the streaming read from Delta (DLT table or any Delta path)
# MAGIC - Use **table FQN** if the table is registered in Unity Catalog.
# MAGIC - Make sure **the producer table is streaming** (append mode) or you use **`readStream`** with incremental changes.

# COMMAND ----------

# Fail fast if table missing
spark.sql(f"DESCRIBE TABLE {source_table_fqn}").collect()

source_stream_df = (
    spark.readStream
         .table(source_table_fqn)       # or .format("delta").load("<path>")
         .withColumn("_eh_enqueued_ts", lit(None).cast("timestamp")) # example: add optional marker field
)

display(source_stream_df, streamName="preview_input_stream")

# COMMAND ----------

# MAGIC %md
# MAGIC ## Start the streaming query
# MAGIC - `checkpointLocation` ensures exactly-once semantics with Delta source.
# MAGIC - `outputMode` is `update` or `append` (no sink here; foreachBatch handles I/O).

# COMMAND ----------

query: StreamingQuery = (
    source_stream_df.writeStream
        .outputMode("update")
        .option("checkpointLocation", checkpoint_path)
        .foreachBatch(foreach_batch_writer)
        .queryName("delta_to_eventhubs_mi")
        .trigger(processingTime="30 seconds")  # tune to your latency/throughput target
        .start()
)

print("Streaming started:", query.name)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Monitoring utilities

# COMMAND ----------

def show_status(q: StreamingQuery):
    print("isActive:", q.isActive)
    print("lastProgress:", q.lastProgress)

show_status(query)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Stopping the stream (when needed)

# COMMAND ----------
# query.stop()
# print("Stopped.")

Notes & tuning

• If using User‑Assigned MI, also set a workspace or cluster environment variable: AZURE_CLIENT_ID=<UAMI client id> (we already pass it explicitly; either works).
• Batch size is auto‑managed by EventDataBatch. If you need strict limits, add a counter before try_add.
• For very high throughput, consider partition routing (assign partition_key) or parallelize by multiple Event Hubs.

Notebook B — DLT pipeline that produces the streaming Delta table

Save as: notebooks/dlt/bronze_silver_orders_dlt.py

This sample creates a small DLT pipeline (Autoloader as example) that writes a streaming Silver table consumed by Notebook A.

# Databricks notebook source
# MAGIC %md
# MAGIC # DLT: Bronze → Silver (Streaming)
# MAGIC Demonstrates a simple streaming DLT pipeline whose output table is read by the Event Hubs writer job.

import dlt
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# Schema example for incoming JSON orders (replace with yours)
order_schema = StructType([
    StructField("order_id",     StringType()),
    StructField("customer_id",  StringType()),
    StructField("amount",       DoubleType()),
    StructField("ts",           TimestampType())
])

@dlt.view(
    comment="Raw streaming ingestion from landing (Autoloader)."
)
def bronze_raw():
    return (
        spark.readStream.format("cloudFiles")
             .option("cloudFiles.format", "json")
             .option("cloudFiles.inferColumnTypes", "true")
             .load("abfss://landing@<storage-account>.dfs.core.windows.net/orders/")  # CHANGE
    )

@dlt.table(
    comment="Cleaned Bronze (schema enforced).",
    table_properties={"quality":"bronze"}
)
def bronze_clean():
    return (
        dlt.read_stream("bronze_raw")
           .select(from_json(col("value").cast("string"), order_schema).alias("j"))
           .select("j.*")
           .where(col("order_id").isNotNull())
    )

@dlt.table(
    comment="Validated Silver table for downstream streaming.",
    table_properties={"quality":"silver"}
)
def silver_orders():
    return (
        dlt.read_stream("bronze_clean")
           .withColumnRenamed("ts", "event_time")
           .where(col("amount") > 0.0)
    )

After publishing this pipeline, Notebook A should point its catalog.schema.table to the DLT target table (e.g., main.stream.silver_orders).

Create & Configure the DLT Pipeline (UI steps)

1. Workflows → Delta Live Tables → Create Pipeline.
2. Development Mode: Start with Continuous (for near‑real time) or Triggered (scheduled).
3. Channel: CURRENT or PREVIEW (use CURRENT for stability).
4. Add notebooks: Library 1: notebooks/dlt/bronze_silver_orders_dlt.py
5. Pipeline settings:
   • Pipeline name: dlt_orders_streaming
   • Storage path: dbfs:/pipelines/dlt_orders_streaming (checkpoints/metadata)
   • Target catalog.schema: e.g., main.stream
   • Compute: Choose a DL workload (enable Photon if available).
6. Permissions: Grant your team CAN VIEW/RUN and data access to the source/target locations.
7. Start pipeline and verify the silver_orders table is created and receiving records.

You do not push directly to Event Hubs from DLT. Keep the sink separation: DLT produces Delta; Notebook A consumes and publishes.

What to capture for stakeholders

1. Databricks Workspace → Identity /Azure Portal Show System‑assigned MI: On (or User‑assigned attached).
   Highlight Object (principal) ID.
2. Event Hubs Namespace → Access control (IAM) → Role assignments Show role: Azure Event Hubs Data Sender assigned to the Databricks MI.
3. Event Hub (entity) → Overview Show Name, Partitions, Throughput units / Premium CU.
4. Workspace → Repos/Notebooks / Databricks Show Notebook A and Notebook B locations.
5. DLT Pipeline Details Pipeline name, Library (Notebook B), Target, Storage path, Compute, Mode.
6. DLT Graph (after start) Nodes: bronze_raw → bronze_clean → silver_orders with green status.
7. Streaming Job (Notebook A) → Spark UI → Structured Streaming Show the running query delta_to_eventhubs_mi.
8. MLflow/Logs panel (optional)
   Show logs indicating batches sent to Event Hubs.
9. Metrics (Event Hubs (Validation)) Incoming Messages, Incoming Requests, Throttled Requests graphs increasing.
10. Process data consumer (optional)

If you have a consumer, show it receiving messages.

Note:

• 401/403 unauthorized: Verify IAM role Event Hubs Data Sender for the same tenant/subscription; if user‑assigned MI, confirm correct client ID.
• ManagedIdentityCredential: Timeout connecting to IMDS: Check cluster outbound to IMDS/managed identity endpoints; ensure workspace MI is enabled.
• Batch too large (EventDataBatch is already full): The code already flushes and re‑creates batches. If frequent, reduce record size or compress payload (e.g., concise JSON).
• Slow throughput: Increase parallelism (more source partitions), add partition keys, increase Event Hubs TU/CU, lower trigger interval (e.g., 10s), or run multiple writers per hub/partition.
• Exactly‑once semantics: With Delta source + checkpoints, duplicates are unlikely; if your sink consumer requires idempotency, include a business key and deduplicate downstream.
• No connection strings in code; AAD MI handles auth.
• Limit MI scope to the minimum (sender on required namespaces only).
• Monitor EH throttling; scale TU/CU or add Kafka‑head client if needed.
• Add dead‑letter pattern if some events fail serialization.

References:

• Structured Streaming with Delta Lake https://learn.microsoft.com/en-us/azure/databricks/structured-streaming/delta-lake
• Azure Event Hubs Python SDK https://learn.microsoft.com/en-us/python/api/overview/azure/eventhub-readme
• Delta Live Tables concepts https://learn.microsoft.com/en-us/azure/databricks/delta-live-tables/
• Implement streaming patterns with DLT https://learn.microsoft.com/en-us/training/modules/implement-streaming-architecture-patterns-with-delta-live-tables/