read_kafka table-valued function

Applies to: Databricks SQL Databricks Runtime 13.3 LTS and above

Reads data from an Apache Kafka cluster and returns the data in tabular form.

Can read data from one or more Kafka topics. It supports both batch queries and streaming ingestion.

Syntax

read_kafka([option_key => option_value ] [, ...])

Arguments

This function requires named parameter invocation.

• option_key: The name of the option to configure. You must use backticks (`) for options that contain dots (.).
• option_value: A constant expression to set the option. Accepts literals and scalar functions.

Returns

Records read from an Apache Kafka cluster with the following schema:

• key BINARY: The key of the Kafka record.
• value BINARY NOT NULL: The value of the Kafka record.
• topic STRING NOT NULL: The name of the Kafka topic the record is read from.
• partition INT NOT NULL: The ID of the Kafka partition the record is read from.
• offset BIGINT NOT NULL: The offset number of the record in the Kafka TopicPartition.
• timestamp TIMESTAMP NOT NULL: A timestamp value for the record. The timestampType column defines what this timestamp corresponds to.
• timestampType INTEGER NOT NULL: The type of the timestamp specified in the timestamp column.
• headers ARRAY<STRUCT<key: STRING, VALUE: BINARY>>: Header values provided as part of the record (if enabled).

Examples

-- A batch query to read from a topic.
> SELECT value::string as value
  FROM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'events'
  ) LIMIT 10;

-- A more advanced query with security credentials for Kafka.
> SELECT * FROM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'events',
    startingOffsets => 'earliest',
    `kafka.security.protocol` => 'SASL_SSL',
    `kafka.sasl.mechanism` => 'PLAIN',
    `kafka.sasl.jaas.config` =>  'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{USER_NAME}" password="{PASSWORD}";',
  );

-- Streaming ingestion from Kafka with JSON parsing.
> CREATE OR REFRESH STREAMING TABLE catalog.schema.raw_events AS
  SELECT
    value::string:events,                 -- extract the field `events`
    to_timestamp(value::string:ts) as ts  -- extract the field `ts` and cast to timestamp
  FROM STREAM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'events'
  );

Options

You can find a detailed list of options in the Apache Spark documentation.

Required options

Provide the option below for connecting to your Kafka cluster.

Provide only one of the options below to configure which Kafka topics to pull data from.

Miscellaneous options

read_kafka can be used in batch queries and in streaming queries. The options below specify which type of query they apply to.

Related articles

• CREATE STREAMING TABLE
• read_files table-valued function