メインコンテンツまでスキップ
に最終更新

Apache Kafka からデータを取り込みます

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

このページでは、Databricks Lakeflow Connect を使用して、マネージド Kafka インジェスト パイプラインを作成する方法について説明します。

要件​

  • 取り込みパイプラインを作成するには、まず、次の要件を満たす必要があります。

    • ワークスペースでUnity Catalogが有効になっている必要があります。

    • サーバレス コンピュートがワークスペースで有効になっている必要があります。サーバレス コンピュートの要件を参照してください。

    • 新しい接続を作成するには、メタストアに対するCREATE CONNECTIONの特権が必要です。Unity Catalogでの特権の管理を参照してください。

      コネクタが UI ベースのパイプライン オーサリングをサポートしている場合、管理者はこのページのステップを完了することで、接続とパイプラインを同時に作成できます。 ただし、パイプラインを作成するユーザーが API ベースのパイプライン オーサリングを使用している場合、または管理者以外のユーザーである場合、管理者はまずカタログ エクスプローラーで接続を作成する必要があります。 「管理対象取り込みソースへの接続」を参照してください。

    • 既存の接続を使用するには、接続オブジェクトに対するUSE CONNECTION権限またはALL PRIVILEGESを持っている必要があります。

    • ターゲットカタログに対するUSE CATALOG権限が必要です。

    • 既存のスキーマに対するUSE SCHEMAおよびCREATE TABLE権限、またはターゲットカタログに対するCREATE SCHEMA権限を持っている必要があります。

  • Kafka から取り込むには、まず「マネージド取り込みのために Apache Kafka に接続する」のステップを完了する必要があります。

取り込みパイプラインを作成​

各Kafkaトピックはストリーミングテーブルに取り込まれます。サポートされているデータと制限については、サポートされているデータを参照してください。

注記

UIベースのパイプラインオーサリングは、ベータ版のKafkaコネクタではサポートされていません。宣言型オートメーションバンドル、またはDatabricksノートブックを使用してパイプラインを作成します。

Kafkaパイプラインをコードとして管理するために、宣言型オートメーションバンドルを使用します。バンドルには、ジョブとタスクの YAML 定義を含めることができ、Databricks CLI を使用して管理され、異なるターゲット ワークスペース（開発、ステージング、本番運用など）で共有および実行できます。詳細については、「宣言型オートメーションバンドルとは」をご覧ください。

  1. Databricks CLI を使用した新しいバンドルの作成

    Bash
    databricks bundle init

  2. バンドルにパイプライン定義ファイルを追加してください (たとえば、resources/kafka_pipeline.yml)。「パイプライン.ingestion_definition」を参照してください。と例

  3. Databricks CLI を使用してバンドルをデプロイする:

    Bash
    databricks bundle deploy

例​

これらの例を使用してパイプラインを構成してください。

最小限のパイプライン — 生のバイナリキーと値​

この例では、キーと値の列をBINARYとして保持した状態で、1つ以上のKafkaトピックを取り込みます。

YAML
variables:
  connection_name:
    default: my-kafka-connection
  dest_catalog:
    default: main
  dest_schema:
    default: kafka_ingest
resources:
  pipelines:
    kafka_pipeline:
      name: kafka-ingestion-pipeline
      serverless: true
      continuous: true
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      target: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        objects:
          - table:
              source_table: N/A
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              destination_table: user_events
              connector_options:
                kafka_options:
                  topics: [user-events, power-user-events]

トランスフォーマー付きパイプライン — JSON値と文字列キー​

この例では、スキーマ進化が有効な状態で、メッセージキーをSTRINGとして、値をJSONとして逆シリアル化します。

YAML
variables:
  connection_name:
    default: my-kafka-connection
  dest_catalog:
    default: main
  dest_schema:
    default: kafka_ingest
resources:
  pipelines:
    kafka_pipeline:
      name: kafka-ingestion-pipeline
      serverless: true
      continuous: true
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      target: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        objects:
          - table:
              source_table: N/A
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              destination_table: user_events
              connector_options:
                kafka_options:
                  topics:
                    - user-events
                  starting_offset: latest
                  key_transformer:
                    format: STRING
                  value_transformer:
                    format: JSON
                    json_options:
                      schema_evolution_mode: rescue

複数のテーブルへのレコードのルーティング (ファンアウト)​

備考

プレビュー

この機能はプライベート プレビュー段階です。試用については、Databricksの担当者にお問い合わせください。

ファンアウトは、単一のKafkaソースから各レコードを多くの宛先テーブルのいずれかにルーティングします。テーブルオブジェクトではなく、スキーマオブジェクトでファンアウトを構成します。各レコードから導出されたルーティングキーによって宛先テーブル名が決定されます：{destination_catalog}.{destination_schema}.{key_value}。

スキーマオブジェクトに fanout_options を設定します。fanout_by フィールドは、生のソースレコード（Kafka の key および value 列を含む）に対して評価される SQL 式であり、その結果が宛先テーブル {destination_catalog}.{destination_schema}.{result} の最終セグメント（テーブル名）になります。fanout_by は生のレコードで実行されるため、次の例のように、フィールドを抽出する前にバイナリ value を文字列にキャストします。解決された値は、引用符で囲んだりサニタイズしたりすることなく、テーブル名のセグメントとしてそのまま使用されるため、有効な引用符で囲まれていないテーブル識別子である必要があります。スペース、ドット、または引用符で囲まれていない識別子で無効なその他の文字を含む値は書き込みが失敗します。また、全て数字である値（例：123）も同様です。値に文字またはアンダースコアも含まれている場合（例：2024_events）、先頭の数字は許可されます。

オプションで、各ルートに単一の JSON 変換を適用できます。この変換はルーティング後、ルーティングされた各レコードで実行されるため、fanout_by が認識する値には影響しません。オプションの完全なリストと v1 の制約については、ファンアウトオプションとファンアウトリミテーションを参照してください。

次の例では、パターンに一致するトピックからレコードを読み込み、各レコードをその event_type フィールドにちなんで名付けられたテーブルにルーティングします。オプションの JSON 変換は、各宛先テーブルでメッセージ value 列をその場でパースします：

YAML
variables:
  connection_name:
    default: my-kafka-connection
  dest_catalog:
    default: main
  dest_schema:
    default: kafka_ingest
resources:
  pipelines:
    kafka_pipeline:
      name: kafka-ingestion-pipeline
      serverless: true
      continuous: true
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      target: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        objects:
          - schema:
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              connector_options:
                kafka_options:
                  topic_pattern: 'events-.*'
                  starting_offset: earliest
              fanout_options:
                fanout_by: 'cast(value as string):event_type::string'
                transforms:
                  - format: JSON
                    input_column: value

一般的なパターン​

高度なパイプライン構成については、「マネージド取り込みパイプラインの一般的なパターン」を参照してください。

次のステップ​

パイプラインでアラートを開始および設定します。共通パイプラインメンテナンス タスクを参照してください。

その他のリソース​

このページの見出し