Note
Access to this page requires authorization. You can try signing in or changing directories.
Access to this page requires authorization. You can try changing directories.
This page contains reference documentation for the managed Kafka connector in Lakeflow Connect.
Connector options
The following options configure the Kafka source for each destination table in the ingestion pipeline. Specify these options under connector_options.kafka_options in your pipeline definition. See Examples for full pipeline examples.
| Option | Type | Default | Description |
|---|---|---|---|
topics |
List of strings | — | List of topic names to subscribe to. Mutually exclusive with topic_pattern. Either topics or topic_pattern is required. |
topic_pattern |
String | — | Java regular expression matching topic names to subscribe to. Mutually exclusive with topics. |
starting_offset |
String | latest |
Where to begin reading when no checkpoint exists (first run only). Valid values: latest, earliest. |
key_transformer |
Transformer | — | Deserializer configuration for message keys. If not set, the key column is retained as BINARY. See Transformer options. |
value_transformer |
Transformer | — | Deserializer configuration for message values. If not set, the value column is retained as BINARY. See Transformer options. |
Transformer options
Transformers define how binary Kafka message keys and values are deserialized into structured columns. Specify the serialization format and the corresponding format-specific options under key_transformer or value_transformer. You can configure a transformer for the key, the value, or both independently. If no transformer is set, the column is retained as BINARY.
For JSON, you can provide an explicit schema, use schema inference with evolution, or omit json_options entirely to store the value as a VARIANT column.
| Option | Applies to | Type | Default | Description |
|---|---|---|---|---|
format |
All | String | — | Serialization format of the data. Valid values: STRING, JSON. STRING requires no additional options. If no json_options are specified on the transformer, the value is parsed as Variant by default. See Variant Data Format for more information. |
json_options.schema |
JSON | String | — | Inline schema in Spark DDL format (for example, "id BIGINT, name STRING"). Mutually exclusive with schema_file_path. |
json_options.schema_file_path |
JSON | String | — | Path to a .ddl schema file. Mutually exclusive with schema. Supports Unity Catalog Volumes paths (/Volumes/...). |
json_options.schema_evolution_mode |
JSON | String | — | Schema evolution mode for automatic schema inference. See Schema Evolution Modes. |
json_options.schema_hints |
JSON | String | — | Comma-separated "column_name type" pairs to influence schema inference (for example, "id BIGINT, ts TIMESTAMP"). Requires schema_evolution_mode to be set. See Override schema inference using schema hints. |
Connection properties
When you create the Unity Catalog Kafka connection in Catalog Explorer, you must specify the following properties depending on the authentication method. See Create a Kafka connection for connection creation steps.
Username and password
| Property | Description |
|---|---|
| Connection name | A unique name for the connection in Unity Catalog. |
| Connection type | Select Kafka. |
| Auth type | Select Username and Password. |
| Username | The username used to authenticate with the Kafka cluster. |
| Password | The password used to authenticate with the Kafka cluster. |
| Bootstrap servers | The bootstrap server address of the Kafka cluster (for example, broker1:9092,broker2:9092). |
| Schema registry URL (optional) | The URL of your schema registry. |
| Schema registry API key (optional) | The API key for your schema registry. |
| Schema registry API secret (optional) | The API secret for your schema registry. |
Service Credential
| Property | Description |
|---|---|
| Connection name | A unique name for the connection in Unity Catalog. |
| Connection type | Select Kafka. |
| Auth type | Select Service Credential. |
| Service credential | Select an existing Unity Catalog service credential or click Create new service credential. |
| Bootstrap servers | The bootstrap server address of the Kafka cluster (for example, broker1:9092,broker2:9092). |
| Schema registry URL (optional) | The URL of your schema registry. |
| Schema registry API key (optional) | The API key for your schema registry. |
| Schema registry API secret (optional) | The API secret for your schema registry. |
Destination table schema
The Kafka connector writes to streaming tables (append-only). The columns written to the destination table depend on whether transformers are configured.
Without transformers (raw binary)
When no key_transformer or value_transformer is configured, the destination table contains the following columns:
| Column | Type | Description |
|---|---|---|
key |
BINARY |
The raw binary content of the Kafka message key. |
value |
BINARY |
The raw binary content of the Kafka message value. |
With a STRING transformer
When format: STRING is set on a transformer, the corresponding column is written as STRING instead of BINARY.
With a JSON transformer
When format: JSON is set on a transformer:
- If no
json_optionsare specified, the column is written asVARIANT. - If
json_options.schemaorjson_options.schema_file_pathis specified, the JSON is parsed into typed columns matching the schema. - If
json_options.schema_evolution_modeis set, schema inference is used and the schema evolves automatically.
Note
Kafka metadata columns — including topic, partition, offset, timestamp, timestampType, and headers — are not available in the destination table in Beta.