read_kafka 表值函数

适用于： Databricks SQL Databricks Runtime 13.3 LTS 及更高版本

从 Apache Kafka 群集读取数据，并以表格形式返回数据。

可从一个或多个 Kafka 主题中读取数据。 支持批处理查询和流式处理引入。

语法

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

参数

此函数需要命名参数调用。

• option_key：要配置的选项的名称。 对于包含点 (.) 的选项必须使用反引号。
• option_value：用于设置选项的常数表达式。 接受文本和标量函数。

返回

记录从使用以下架构的 Apache Kafka 群集中读取的内容：

• key BINARY：Kafka 记录的密钥。
• value BINARY NOT NULL：Kafka 记录的值。
• topic STRING NOT NULL：从其中读取记录的 Kafka 主题的名称。
• partition INT NOT NULL：从其中读取记录的 Kafka 分区的 ID。
• offset BIGINT NOT NULL：Kafka“TopicPartition”中记录的偏移量。
• timestamp TIMESTAMP NOT NULL：记录的时间戳值。 “timestampType”列定义此时间戳对应的内容。
• timestampType INTEGER NOT NULL：“timestamp”列中指定的时间戳的类型。
• headers ARRAY<STRUCT<key: STRING, VALUE: BINARY>>：作为记录一部分而提供的标头值（如已启用）。

示例

-- 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'
  );

选项

可以在 Apache Spark 文档中查看详细的选项列表。

必需选项

提供以下用于连接到 Kafka 群集的选项。

仅提供以下其中一个选项来配置从哪个 Kafka 主题拉取数据。

杂项选项

read_kafka 可以在批处理查询和流式处理查询中使用。 以下选项指定它们适用于的查询类型。

相关文章

• CREATE STREAMING TABLE
• read_files 表值函数