Lakeflow Connect: Message Bus Ingestion - Now shipping your logs directly! (Beta)

[Vicky_Bukta_DB]

Take a quick inventory of where your operational data lives right now. There is probably a Kafka cluster carrying your event backbone, or perhaps RabbitMQ capturing messages on a factory floor or between services. Getting this data into your lakehouse for analytics has traditionally meant standing up, scaling, and babysitting a custom ingestion pipeline for every topic or queue.

Lakeflow Connect now eliminates this maintenance burden. Kafka and RabbitMQ ingestion are now available as fully managed streaming connectors (Beta),  allowing you to bring your most demanding operational streams into the lakehouse with zero infrastructure to manage. You simply point a connector at your source stream, and the data flows continuously into Delta.

Your workload changes, your pipelines should too

Lakeflow Connect for message buses is packed with features to handle your toughest workloads. Advanced functionality, such as bin-packing under the hood, allows us to effectively distribute and bundle your topic ingestion across compute resources for optimal performance. 

Think back to how you managed your low- to moderate-workload. How do they change and fluctuate over the course of a week or month? It's common to see events spike during peak periods, such as flash sales, working hours, or newsletters being sent out. This translates to fluctuations in volume for your topics. Having separate dedicated pipelines is great for durability, but can be costly. On the other hand, bundling your topics into a single pipeline can be messy to maintain when scaling clusters is required for matriculation.

Bin-packing allows managed Lakeflow Connect streaming ingestion to use a single CPU core to process data from multiple Kafka partitions within a single task more efficiently. Lakeflow Connect is also serverless, which means that you only pay for the compute you use, rather than the compute you provisioned. The result is a cost-effective pipeline that can scale elasticity in response to your changing volumes.

Getting started with the managed connectors

These managed connectors are configured today with Declarative Automation Bundles (DABs), with the exception of RabbitMQ, which is coming soon, and the pipeline API, which means you can author them from a notebook or check them into your source code. Point-and-click UI authoring is coming.

Each connector uses two pieces:

1. A Connection, a Unity Catalog securable that stores the broker endpoint and credentials. You create it once in Catalog Explorer (choose the connector type and enter the bootstrap servers or broker endpoint plus authentication), and reference it by name in your pipelines.
2. An ingestion pipeline, defined as a bundle, that maps sources to destination tables and runs on serverless compute.

Example

Here is a minimal Kafka ingestion pipeline. It reads two topics into a single destination table.

This example demonstrates the following:

1. Selecting the UC connection to our Kafka cluster via connection_name.
2. Describing the managed Kafka pipeline via kafka_pipeline
1. Choosing a destination via destination_catalog, destination_schema, which represents the schema, and destination_table. Note that fanout rules can be specified to autocreate tables.

variables:
  connection_name:
    default: my-kafka-connection
  dest_catalog:
    default: main
  dest_schema:
    default: kafka_ingest
resources:
  pipelines:
    kafka_pipeline:
      name: kafka-ingestion-pipeline
      continuous: true
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      target: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        objects:
          - table:
              source_table: N/A
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              destination_table: user_events
              connector_options:
                kafka_options:
                  topics: [user-events, power-user-events]

Deploy it with the bundle CLI:

databricks bundle init
databricks bundle deploy

The connector handles the rest: reading continuously from the topics and writing to the user_events streaming table.

Flexible data handling with Variant

 Kafka messages are rarely perfectly uniform, so the connector lets you say how to deserialize values and what to do when the shape drifts. This is where it connects to schema management directly.

This example demonstrates the following:

1. format is set to STRING and JSON. The JSON body of the message will then be parsed to a variant column for ultimate flexibility.
2. Starting_offset is set to the latest signal from which we should start ingestion in the topic. Options are the latest and the earliest. Learn more from our documentation here.

ingestion_definition:
  connection_name: ${var.connection_name}
  objects:
    - table:
	 source_table: N/A
        destination_catalog: ${var.dest_catalog}
        destination_schema: ${var.dest_schema}
        destination_table: user_events
        connector_options:
          kafka_options:
            topics:
              - user-events
            starting_offset: latest
            key_transformer:
              format: STRING
            value_transformer:
              format: JSON

Schema Inference and Evolution.

Though schemas drift slowly, teams typically establish rules for schema evolution to ensure backward compatibility (e.g., adding optional columns or widening types). Lakeflow Connect automatically applies forward-rolling schema updates, maintaining schematized data and optimal query performance

ingestion_definition:
  connection_name: ${var.connection_name}
  objects:
    - table:
	 source_table: N/A
        destination_catalog: ${var.dest_catalog}
        destination_schema: ${var.dest_schema}
        destination_table: user_events
        connector_options:
          kafka_options:
            topics:
              - user-events
            starting_offset: latest
            key_transformer:
              format: STRING
            value_transformer:
		format: JSON
		json_options:
		  schema_evolution_mode: ADD_NEW_COLUMNS

Routing to many tables (fanout). 

Other times, it's not about schema drift. It's the fact that a single Kafka topic could contain many different event types. Our fanout capability lets you route events to different tables using an expression, rather than defining a pipeline per table.

objects:
  - schema:
      source_schema: N/A
      destination_catalog: ${var.dest_catalog}
      destination_schema: ${var.dest_schema}
      connector_options:
        kafka_options:
          topic_pattern: 'events-.*'
          starting_offset: earliest
	   value_transformer: 
     format: STRING
      fanout_options:
        fanout_by: 'cast(value as string):event_type::string'
        transforms:
          - format: JSON
            input_column: value

Decision matrix: managed connector or forwarder?

There are two approaches to move your data: Push vs. pull.

Our managed streaming connectors are pull-based. Lakeflow Connect reaches into your source, polls for data, and replicates it. Due to the fact that it reaches in, the connector requires inbound network access to the source. In exchange, there is no software for you to run or maintain. It is a fully managed service.

A Zerobus Ingest forwarder is push-based. A sidecar process you build with the Zerobus SDK runs inside your perimeter and pushes data out, so all traffic is outbound only. Zerobus is flexible enough that you can build a forwarder from almost any source system. The cost is that you own and run that agent. 

Our connector and Zerobus Ingest services are therefore complementary, rather than competing.  Evaluate the tradeoff between the managed connector and the Zerobus forwarder based on your security posture and operational constraints, as outlined in the matrix below.

	Managed streaming connector	Zerobus forwarder
Direction	Pull. Lakeflow Connect replicates from the source	Push. Your agent pushes records out
Who runs it	Fully managed by Lakeflow Connect. No software to run	You build and run the agent (using the Zerobus SDK)
Network	Requires inbound access to the source	Outbound only from inside your perimeter
Source of truth	Stays in the message bus	Is the lakehouse.
Sources	Kafka, RabbitMQ, other coming soon.	Any source you can write code against.

What's next?

The managed streaming connectors are just getting started. UI-based authoring is on the way and more data sources are coming soon. On the horizon, we are adding native support for Avro and Protobuf, including full schema registry integration. Full UI experience is around the corner.

Zerobus Ingest is adding Apache Kafka-compatible and MQTT-compatible producer APIs for the push-based side of the story. Whichever direction your data needs to move, the trade-offs in the matrix above will still guide you.

Ready to try it? Check out the Kafka and RabbitMQ connector docs to set up your first managed pipeline.

Have questions, or want to share which message buses you are bringing into your lakehouse? Join the discussion below!