read_kafka テーブル値関数

適用対象: Databricks SQL Databricks Runtime 13.3 LTS 以上

Apache Kafka クラスターからデータを読み取り、表形式でデータを返します。

1 つ以上の Kafka トピックからデータを読み取ることができます。 バッチ クエリとストリーミング インジェストの両方がサポートされています。

構文

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

引数

この関数には、名前付きパラメーター呼び出しが必要です。

• option_key: 構成するオプションの名前。 バックティック () for options that contain dots (.`) を使用する必要があります。
• 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;

-- 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 テーブル値関数
• read_pubsub ストリーミング テーブル値関数