Função com valor de tabela read_kafka

Aplica-se a: SQL do Databricks Databricks Runtime 13.3 LTS e versões posteriores

Lê dados de um cluster do Apache Kafka e retorna os dados em formato tabular.

Pode ler dados de um ou mais tópicos do Kafka. É compatível com consultas em lote e ingestão de streaming.

Sintaxe

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

Argumentos

Essa função requer invocação de parâmetro nomeada.

• option_key: o nome da opção a ser configurada. Você deve usar acento grave (`) para opções que contenham pontos (.).
• option_value: uma expressão constante para definir a opção. Aceita literais e funções escalares.

Retornos

Registros lidos de um cluster do Apache Kafka com o seguinte esquema:

• key BINARY: a chave do registro Kafka.
• value BINARY NOT NULL: o valor do registro Kafka.
• topic STRING NOT NULL: o nome do tópico Kafka do qual o registro é lido.
• partition INT NOT NULL: a ID da partição Kafka da qual o registro é lido.
• offset BIGINT NOT NULL: o número de deslocamento do registro no Kafka TopicPartition.
• timestamp TIMESTAMP NOT NULL: um valor de carimbo de data/hora para o registro. A coluna timestampType define ao que esse carimbo de data/hora corresponde.
• timestampType INTEGER NOT NULL: o tipo do carimbo de data/hora especificado na coluna timestamp.
• headers ARRAY<STRUCT<key: STRING, VALUE: BINARY>>: valores de cabeçalho fornecidos como parte do registro (se habilitado).

Exemplos

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

Opções

Você pode encontrar uma lista detalhada de opções na Documentação do Apache Spark.

Opções necessárias

Forneça a opção abaixo para se conectar ao cluster do Kafka.

Forneça apenas uma das opções abaixo para configurar quais tópicos do Kafka serão usados para efetuar pull de dados.

Opções diversas

read_kafka podem ser usados nas consultas em lote e nas consultas de streaming. As opções a seguir especificam a que tipo de consulta elas se aplicam.

Artigos relacionados

• CREATE STREAMING TABLE
• Função read_files com valor de tabela
• função com valor de tabela de streaming read_pubsub