Ingest a streaming Kafka topic¶

Your order events don’t wait for a nightly batch — they land on a Kafka topic the moment each one happens. To get them into the lakehouse you need a continuous reader: connect to the brokers, subscribe to the topic, authenticate to a secured cluster, and keep streaming.

You could hand-write that reader — spark.readStream.format("kafka") with a stack of .option(...) calls, a dbutils.secrets.get(...) for the truststore password, then a create_streaming_table and an append_flow to land the rows. Or you declare a type: kafka load and let Lakehouse Plumber write that Lakeflow for you. That is the idea on every page: declare your ETL, don’t hand-write it.

Let’s stream an orders topic off a TLS-secured Kafka cluster into a bronze streaming table, without touching PySpark.

Before you start¶

You need Lakehouse Plumber installed and a project to work in. If you followed the Get Started course you already have one; otherwise scaffold an empty project first. This guide assumes the orders topic already exists on your Kafka cluster, and that the truststore password is stored in a Databricks secret scope named kafka — a Kafka load reads a live topic, it does not create it.

Declare the Kafka load¶

A pipeline in Lakehouse Plumber is a flowgroup: a short YAML file describing a sequence of actions. This one has two — a load that streams the topic into a view, and a write that appends the view into a streaming table.

Create pipelines/orders_stream.yaml:

pipelines/orders_stream.yaml¶

# Ingest guide: stream a Kafka topic into a bronze streaming table.
# Action chain: load (kafka) -> write (streaming_table).
pipeline: bronze_ingest
flowgroup: orders_stream

actions:
  # 1. Load: stream the "orders" topic from a TLS-secured Kafka cluster.
  #    Kafka is always a streaming source; the truststore password comes
  #    from a Databricks secret rather than living in the YAML.
  - name: load_orders_kafka
    type: load
    source:
      type: kafka
      bootstrap_servers: "${kafka_bootstrap}"
      subscribe: orders
      options:
        startingOffsets: earliest
        kafka.security.protocol: SSL
        kafka.ssl.truststore.location: "${truststore_path}"
        kafka.ssl.truststore.password: "${secret:kafka/truststore-password}"
    target: v_orders_kafka_raw

  # 2. Write: append the raw Kafka envelope into a bronze streaming table.
  - name: write_orders_bronze
    type: write
    source: v_orders_kafka_raw
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${bronze_schema}"
      table: orders_raw

The load action names the cluster with bootstrap_servers and picks the topic with subscribe. Kafka is always a streaming source, so there is no batch alternative: LHP generates spark.readStream.format("kafka") and readMode defaults to stream. Everything under options is passed straight to the Kafka reader — startingOffsets: earliest reads the topic from the beginning on the first run, and the three kafka.* keys turn on TLS and point at the truststore.

The truststore password is a secret, not a literal. ${secret:kafka/truststore-password} references the truststore-password key in the kafka secret scope; LHP turns that into a dbutils.secrets.get(...) call at generation time, so the password never appears in your YAML or in the generated Python. The other ${...} tokens are per-environment placeholders you resolve next.

subscribe is one of three subscription methods — the load action reference covers subscribePattern (a topic regex) and assign (specific partitions), the extra kafka.* authentication options for AWS MSK and Azure Event Hubs, and offset controls like failOnDataLoss. This guide stays on the single-topic case.

Point the tokens at your environment¶

The ${...} tokens resolve from a per-environment substitutions file. Create substitutions/dev.yaml:

substitutions/dev.yaml¶

# Development environment tokens.
# Referenced from flowgroups with ${token} syntax.
dev:
  catalog: dev_catalog
  bronze_schema: bronze
  kafka_bootstrap: kafka-dev.internal:9092
  truststore_path: /Volumes/dev_catalog/kafka/truststore.jks

This is the only place environment-specific values live. To stream the same topic from a production cluster later, you write a prod.yaml with the production brokers, truststore, and catalog — the flowgroup itself never changes.

Generate the pipeline¶

Now compile the YAML into Lakeflow code. Validate first, then generate:

$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_ingest  0 files
✓ validate (0.32s)
1 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.00s)
✓ preflight (0.00s)
✓ bronze_ingest  1 file
✓ generate (0.36s)
✓ format (0.10s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.5s

validate resolves every token and secret reference and checks the actions before you commit to generating; generate writes the Python.

Read what Lakehouse Plumber wrote¶

Open generated/dev/bronze_ingest/orders_stream.py. This is the entire output — nothing is hidden behind a runtime:

generated/dev/bronze_ingest/orders_stream.py¶

# Generated by LakehousePlumber
# Pipeline: bronze_ingest
# FlowGroup: orders_stream

from pyspark import pipelines as dp

# Pipeline Configuration
PIPELINE_ID = "bronze_ingest"
FLOWGROUP_ID = "orders_stream"


# ============================================================================
# SOURCE VIEWS
# ============================================================================


@dp.temporary_view()
def v_orders_kafka_raw():
    """Load data from Kafka topics at kafka-dev.internal:9092"""
    df = (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "kafka-dev.internal:9092")
        .option("subscribe", "orders")
        .option("startingOffsets", "earliest")
        .option("kafka.security.protocol", "SSL")
        .option(
            "kafka.ssl.truststore.location", "/Volumes/dev_catalog/kafka/truststore.jks"
        )
        .option(
            "kafka.ssl.truststore.password",
            dbutils.secrets.get(scope="kafka", key="truststore-password"),
        )
        .load()
    )

    return df


# ============================================================================
# TARGET TABLES
# ============================================================================

# Create the streaming table
dp.create_streaming_table(
    name="dev_catalog.bronze.orders_raw", comment="Streaming table: orders_raw"
)


# Define append flow(s)
@dp.append_flow(
    target="dev_catalog.bronze.orders_raw",
    name="f_orders_bronze",
    comment="Append flow to dev_catalog.bronze.orders_raw",
)
def f_orders_bronze():
    """Append flow to dev_catalog.bronze.orders_raw"""
    # Streaming flow
    df = spark.readStream.table("v_orders_kafka_raw")

    return df

It’s ordinary Lakeflow. The source view is a @dp.temporary_view wrapping spark.readStream.format("kafka"), with every option(...) you declared resolved in order and the token values substituted in. The secret became a bare dbutils.secrets.get(scope="kafka", key="truststore-password") call — a live lookup at run time, not a string. The target is a dp.create_streaming_table, and a @dp.append_flow streams the view into it. You described a topic and a target table; LHP wrote the reader, the secret lookup, and the two pieces of plumbing that connect them.

Note

A Kafka read returns a fixed schema: key and value as raw binary columns, plus topic, partition, offset, timestamp, and timestampType. This flowgroup lands that envelope as-is in the bronze orders_raw table — the durable, replayable record of what arrived. To turn the binary value into typed columns, deserialize it downstream with a SQL transform (CAST(value AS STRING) then from_json(...)), covered in the SQL transform guide.

What you just did¶

Twenty-three lines of YAML. Sixty lines of Lakeflow Python. Zero lines of PySpark from you — you never wrote readStream.format("kafka"), the six .option(...) calls, the dbutils.secrets.get lookup, create_streaming_table, or append_flow. And the output is code you own: version it, diff it, and open it in the Databricks editor like anything else.

The payoff compounds. Subscribing to a second topic, switching brokers per environment, or authenticating with MSK IAM is a few more lines of YAML — not another sixty lines of Python each time.

What’s next¶

Deserialize the payload. The binary value column becomes typed rows with a SQL transform — CAST(value AS STRING) and from_json(...) — reading stream(v_orders_kafka_raw). See the SQL transform guide.
Authenticate to a managed broker. AWS MSK IAM and Azure Event Hubs OAuth are extra kafka.sasl.* options on the same load, with credentials supplied as secrets. The load action reference documents both, end to end.
See every Kafka option. subscribePattern and assign, startingOffsets, failOnDataLoss, and the full kafka.* set are in the load action reference, with their constraints.