Set up a Stream

A Stream represents an external streaming data source, such as Apache Kafka. Streams store connection details, authentication, schemas, and ingestion configuration. After a stream is created, you can reference it using Feature View definitions to create real-time streaming features.

Streams have three-part names (catalog.schema.stream_name). Access to a Stream is governed by its associated ingestion table. See Ingestion and backfill for details.

Requirements

• For running notebook commands: serverless or a classic compute cluster running Databricks Runtime 17.0 ML or above.
• The feature-engineering-client Python package version 0.17.0 or above must be installed.

Create a stream

Use create_stream() to create a new Stream. A Stream requires four configuration components:

• Source config: Specifies the streaming platform (for example, Kafka) and source-specific details (such as topic subscription for Kafka).
• Connection config: Specifies how to connect and authenticate to the streaming platform, including bootstrap servers and credentials.
• Schema config: Defines the structure of message keys and values.
• Ingestion config: Specifies where and how stream data is ingested. See Ingestion and backfill for details.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KafkaStreamConfig,
    KafkaSubscriptionMode,
    StreamConnectionConfig,
    DirectSchemas,
    SchemaConfig,
    IngestionConfig,
    IngestionDestination,
    StreamBackfillSource,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KafkaStreamConfig(
        subscription_mode=KafkaSubscriptionMode(subscribe="events-topic"),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kafka-connection"
    ),
    schema_config=DirectSchemas(
        payload_schema=SchemaConfig(
            json_schema=(
                '{'
                '  "type": "object",'
                '  "properties": {'
                '    "transaction_id": {"type": "string"},'
                '    "user_id": {"type": "string"},'
                '    "amount": {"type": "number"},'
                '    "event_time": {"type": "string", "format": "date-time"}'
                '  }'
                '}'
            )
        ),
    ),
    ingestion_config=IngestionConfig(
        ingestion_destination=IngestionDestination(
            delta_table_name="my_catalog.my_schema.events_ingestion"
        ),
    ),
)

Connecting to stream sources

Before defining streaming features, connect and test a streaming Lakeflow pipeline connection to your Kafka broker. See Streaming on serverless compute and Connect to Apache Kafka.

For AWS managed streaming (Amazon MSK), see Serverless private connectivity to Amazon MSK. For details on Kafka authentication options, see Authentication.

Authentication

Unity Catalog connection (recommended)

Use a Unity Catalog connection to authenticate to your Kafka cluster. This is the recommended approach for managed authentication. To create a connection, see Create a connection. The Stream's creator must have USE CONNECTION on the connection. Any user materializing features with the Stream as a source must also have USE CONNECTION on the connection.

connection_config = StreamConnectionConfig(
    uc_connection_name="my-kafka-connection"
)

The connection supports both IAM (service credential) and SASL authentication.

IAM (service credential)

Authenticate with a Unity Catalog service credential, for example to connect to Amazon MSK with IAM. To create a service credential, see Create service credentials. Set the service credential name with the credential option:

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    credential '<service_credential>'
)

In addition to USE CONNECTION on the connection, identities that use the service credential need ACCESS on it. Grant ACCESS on the referenced service credential to the Stream's creator and to any identity that materializes features with the Stream. See Grant permissions to use a service credential to access an external cloud service.

SASL

SASL authentication uses a username and password. Set sasl_mechanism to one of the following:

• PLAIN
• SCRAM-SHA-256
• SCRAM-SHA-512

Provide the credentials with the user and password options. The connection stores these credentials securely.

The following example uses SASL/SCRAM. For SASL/PLAIN, set sasl_mechanism to PLAIN.

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    sasl_mechanism 'SCRAM-SHA-512',
    user '<username>',
    password '<password>'
)

Direct mTLS

For direct mTLS authentication, provide keystore and truststore files stored on a Unity Catalog volume, with passwords referenced through Databricks secret scopes. For more information on SSL authentication with Kafka, see Use SSL to connect Azure Databricks to Kafka.

from databricks.feature_engineering.entities import (
    DirectMtlsConfig,
    MtlsConfig,
    SecretScopeReference,
)

connection_config = DirectMtlsConfig(
    bootstrap_servers="broker1:9092,broker2:9092",
    mtls_config=MtlsConfig(
        keystore_location="/Volumes/my_catalog/my_schema/my_volume/keystore.jks",
        keystore_password_ref=SecretScopeReference(
            scope="my_scope", key="keystore_password"
        ),
        key_password_ref=SecretScopeReference(
            scope="my_scope", key="key_password"
        ),
        truststore_location="/Volumes/my_catalog/my_schema/my_volume/truststore.jks",
        truststore_password_ref=SecretScopeReference(
            scope="my_scope", key="truststore_password"
        ),
    ),
)

Subscription modes

The subscription mode specifies how the Stream selects Kafka topics to consume from. Three modes are supported:

Schema configuration

Define the structure of message keys and values so that ingestion and feature definitions can read individual fields. For Kafka sources, payload_schema corresponds to the Kafka message value (the value in Kafka's key-value model) and key_schema corresponds to the Kafka message key. At least one of payload_schema or key_schema must be provided.

Each SchemaConfig accepts one of three formats, matching how the source serializes its messages: json_schema, avro_schema, or proto_schema. If no schema is provided for a key or payload, it is treated as a simple string.

The code examples in this section use schemas declared inline with DirectSchemas, where the schema is provided as a string. To manage schemas using an external schema registry see Schema registry for details.

JSON schema

Provide a JSON Schema string to json_schema.

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        json_schema=(
            '{'
            '  "type": "object",'
            '  "properties": {'
            '    "user_id": {"type": "string"},'
            '    "amount": {"type": "number"},'
            '    "event_time": {"type": "string"}'
            '  }'
            '}'
        )
    ),
    key_schema=SchemaConfig(
        json_schema='{"type": "string"}'
    ),
)

Avro schema

Provide an Avro schema string to avro_schema. Avro logical types are supported, including timestamp-millis, date, and decimal.

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        avro_schema=(
            '{'
            '  "type": "record",'
            '  "name": "Event",'
            '  "fields": ['
            '    {"name": "user_id", "type": "string"},'
            '    {"name": "amount", "type": "double"},'
            '    {"name": "event_time",'
            '     "type": {"type": "long", "logicalType": "timestamp-millis"}}'
            '  ]'
            '}'
        )
    ),
)

Protobuf schema

Provide a ProtoSchemaSpec to proto_schema with the Protocol Buffers .proto source text and the payload message name. Import ProtoSchemaSpec from databricks.feature_engineering.entities.

message_name must be the fully-qualified message name, including the package declared in the .proto text (for example, com.example.Event, not Event). Both proto2 and proto3 syntax are supported.

google.protobuf.Timestamp and the scalar wrapper types (StringValue, Int32Value, and so on) are supported, and their imports are resolved automatically. Other well-known types, such as Duration, Struct, and Any, are rejected; encode those values as a supported scalar or message instead. The fixed32 and fixed64 scalar types and map with non-string keys are also not supported.

from databricks.feature_engineering.entities import ProtoSchemaSpec

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        proto_schema=ProtoSchemaSpec(
            schema_text=(
                'syntax = "proto3";\n'
                'package com.example;\n'
                'import "google/protobuf/timestamp.proto";\n'
                'message Event {\n'
                '  string user_id = 1;\n'
                '  double amount = 2;\n'
                '  google.protobuf.Timestamp event_time = 3;\n'
                '}'
            ),
            message_name="com.example.Event",
        )
    ),
)

Decoding data using schemas

Databricks decodes each message with Spark's from_json, from_avro, and from_protobuf functions. The following behaviors apply whether you declare the schema inline or resolve it from a schema registry:

• Malformed records. Decoding uses the PERMISSIVE mode, so a record that does not match its schema decodes to a null value instead of failing the stream.
• Avro unions. A union of multiple record types decodes to a struct with one field per record type, each named after its Avro record.
• Protobuf types. Unsigned integers decode to a wider signed type (for example, uint32 to BIGINT and uint64 to DECIMAL(20,0)), enum fields decode to their string name, and scalar wrapper types (for example, StringValue and Int32Value) decode to a nullable column of the wrapped type.

Schema registry

Schema registries store and version schemas that streaming producers and consumers use, enforcing compatibility rules as those schemas evolve. When an external schema registry is configured, Feature Store reads the schema from the registry and uses it to decode the streaming message. You do not declare the schema inline on the Stream when using a schema registry.

Schema registry support has the following limitations:

• Only Confluent Schema Registry is supported
• Only the Avro and Protobuf formats are supported. To read JSON messages, declare the schema inline instead. See JSON schema.
• Each Stream is connected to exactly one Confluent subject for the message value, and one for the message key (if provided). Stream topics containing multiple schema records is not a supported configuration. If your Stream connects to topics that contain multiple schemas, records that do not match the schema for the specified subject are decoded as null.

Connect to a schema registry

Provide the registry connection details as options on the Kafka Unity Catalog connection, and store the registry API secret in a Databricks secret scope. The Stream's run-as identity must have READ permission on the secret scope, because the ingestion pipeline reads the secret at runtime. For how to create and configure a connection, see Create a connection.

Add the schema_registry_url, schema_registry_api_key, and schema_registry_api_secret options to the connection used for authentication. The following example creates a Kafka connection that authenticates to the broker with a Unity Catalog service credential and to the registry with an API key:

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    credential '<service_credential>',
    schema_registry_url 'https://<registry-host>',
    schema_registry_api_key '<registry_api_key>',
    schema_registry_api_secret secret('<scope>', '<key>')
)

Set both the schema_registry_api_secret option on the Kafka connection and the secret scope reference on the Stream to the same secret.

Create a stream that uses a schema registry

Pass a SchemaRegistryConfig as the schema_config. Reference the registry API secret with api_secret_ref, and identify the subject and format with payload_schema_locator for the message value, or key_schema_locator for the message key. At least one locator must be provided.

Note the differences here compared to the direct schema examples in the Schema configuration section. When using a schema registry, you do not provide the schema inline on the Stream to schema_config. Instead, you specify a SchemaRegistryConfig that identifies the schema in the registry.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KafkaStreamConfig,
    KafkaSubscriptionMode,
    StreamConnectionConfig,
    SchemaRegistryConfig,
    SchemaLocator,
    SchemaLocatorConfluentSchema,
    SchemaLocatorFormat,
    SecretScopeReference,
    IngestionConfig,
    IngestionDestination,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KafkaStreamConfig(
        subscription_mode=KafkaSubscriptionMode(subscribe="transactions"),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kafka-connection"
    ),
    schema_config=SchemaRegistryConfig(
        api_secret_ref=SecretScopeReference(
            scope="my_scope", key="sr_api_secret"
        ),
        payload_schema_locator=SchemaLocator(
            confluent_schema=SchemaLocatorConfluentSchema(
                subject="transactions-value"
            ),
            format=SchemaLocatorFormat.FORMAT_AVRO,
        ),
    ),
    ingestion_config=IngestionConfig(
        ingestion_destination=IngestionDestination(
            delta_table_name="my_catalog.my_schema.transactions_ingestion"
        ),
    ),
)

A Confluent subject is the named scope under which a schema's version history is registered and compatibility is enforced. Set subject to the relevant scope's name, which is commonly determined from the subject name strategy:

• TopicNameStrategy (default, derives the subject from the topic name): <topic>-value for the value and <topic>-key for the key. For example, the value schema for the topic transactions uses the subject transactions-value.
• RecordNameStrategy (derives the subject from the schema's record name, independent of the topic): the fully-qualified record name, such as com.example.Payment. This is the record's namespace and name for Avro, or the message's package and name for Protobuf.
• TopicRecordNameStrategy (combines the topic and record names): <topic>-<fully-qualified-record-name>, such as transactions-com.example.Payment.

format is required. Set it to SchemaLocatorFormat.FORMAT_AVRO or SchemaLocatorFormat.FORMAT_PROTOBUF to match how the topic is serialized.

Schema evolution

The ingestion pipeline resolves the subject's current schema when it starts. When you register a new backward-compatible schema version on the subject in the schema registry, the running pipeline continues to use the version it started with.

Because Databricks manages the ingestion pipeline as a serverless Lakeflow pipeline, the pipeline restarts periodically. On its next restart, it picks up the new schema version. It can take up to a week for new or changed fields to appear in the ingestion table.

For how the pipeline handles records that don't match the schema it is currently using, see Decoding data using schemas.

Ingestion and backfill

The ingestion_config parameter configures how stream data is captured and stored for training and serving.

Access to a Stream is governed by the ingestion table:

• SELECT on the ingestion table grants read access to the Stream.
• MANAGE on the ingestion table grants delete access.

For more information on table privileges, see Table and Unity Catalog privileges reference.

Ingestion pipeline

When a stream is created, Databricks starts a managed ingestion pipeline that continuously reads messages from the Kafka topic and writes them into a Delta table (the ingestion table). The pipeline starts from the latest Kafka offset and runs continuously, capturing only new messages that arrive after the stream is created. This ingestion table is used for training with streaming features. When a stream is deleted, its ingestion pipeline and ingestion table are also deleted.

Ingestion destination

The ingestion_destination specifies the three-part Delta table name where stream data is written.

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
)

Ingestion table schema

The ingestion table contains the message data along with metadata columns:

Backfill source

Because the forward-fill pipeline starts from the latest Kafka offset, it does not capture messages that existed before the stream was created. To provide historical data coverage for training, configure an optional backfill source.

When a backfill source is configured, Databricks runs a one-time MERGE INTO job that copies backfill rows into the ingestion table with record_source="backfill". The MERGE runs only after the overlap checker confirms that the backfill source and the forward-fill stream have overlapping timestamps (see Overlap between backfill and live stream data). If the overlap condition is not met within 2 days, the MERGE runs anyway to avoid blocking indefinitely.

The backfill table must include a stream_record_timestamp column of type TIMESTAMP in UTC timezone. Other Kafka metadata columns (kafka_topic, kafka_partition, kafka_offset) are passed through if present on the backfill source, or set to NULL otherwise.

from databricks.feature_engineering.entities import StreamBackfillSource

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
    backfill_source=StreamBackfillSource(
        delta_table_name="my_catalog.my_schema.historical_events"
    ),
)

Overlap between backfill and live stream data

Before running a MERGE between backfill and the ingestion table, an overlap check compares timestamps on the two tables:

• Backfill max: The maximum stream_record_timestamp in the backfill source.
• Ingestion min: The minimum stream_record_timestamp of rows (record_source="stream") in the ingestion table.

The MERGE proceeds when the backfill's latest timestamp exceeds the ingestion table's earliest timestamp by at least 1 hour. This overlap ensures there are no gaps in the ingestion table. If the overlap condition is not met within 2 days, the MERGE runs anyway to avoid blocking indefinitely.

Because the ingestion pipeline starts from the latest Kafka offset, it only captures messages arriving after the stream is created. Your backfill source must contain data that extends into the ingestion time range — not just up to the stream creation time.

For example, if you create a stream at 3:00 PM, the forward-fill pipeline begins reading messages from 3:00 PM onward. Your backfill source must include data with timestamps through at least 4:00 PM (1 hour past the forward-fill start) to satisfy the overlap check. This means you should update your backfill table after 4:00 pm to ensure ingestion table has no gaps.

Deduplication

Use deduplication_columns to specify column paths for identifying duplicate rows during ingestion between backfill and forward-fill stream data. Use dot notation for nested fields (for example, "value.user_id").

Choose deduplication columns based on your data:

• If each record in your stream contains a unique identifier (for example, value.transaction_id), use that column for deduplication.
• If your backfill source includes kafka_partition and kafka_offset columns, use those to uniquely identify each record.
• If no deduplication columns are specified, the default deduplication key is the full combination of key, value, and stream_record_timestamp. This is not recommended as this strict criteria matching can easily lead to duplicates.

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
    deduplication_columns=["value.transaction_id"],
)

Manage streams

Get a stream

stream = client.get_stream(name="my_catalog.my_schema.my_stream")

List streams

streams = client.list_streams(
    catalog_name="my_catalog",
    schema_name="my_schema",
    max_results=50,
    include_schemas=False,
)

Set include_schemas=True to include full schema details. Schemas can be large and this might result in a long-running operation. To retrieve schemas individually instead, use get_stream.

Delete a stream

Deleting a stream also deletes its ingestion pipeline and ingestion table.

client.delete_stream(name="my_catalog.my_schema.my_stream")

Example notebook

For an end-to-end example that creates a Stream, defines streaming features, and deploys to a serving endpoint, see the following notebook:

Streaming Feature Views quickstart notebook

Get notebook