Zerobus IngestではKafka互換APIを使用してください

Zerobus Ingestは、Azure Databricks SDKを使わず、任意のApache Kafkaプロデューサークライアントを使って取り込むことができるKafka互換のプロデューサーAPIを提供しています。 既存のKafkaプロデューサーをZerobusエンドポイントに向け、ターゲットテーブルの名前が付けられたトピックに出力すると、そのレコードはUnity CatalogのDeltaテーブルに格納されます。 Kafka互換APIは、すでにKafkaプロデューサーやKafkaプロトコルを話すコレクタ、あるいはKafkaに出力するツールがあり、最小限のコード変更でそのデータをDeltaにルーティングしたい場合に適しています。

Kafka互換APIはTCP上で SASL_SSL と OAUTHBEARER メカニズムを用いて動作し、 Produce、 Metadata、 ApiVersions、SASLのハンドシェイクAPIを含むKafkaプロトコルのプロデューサー側サブセットを実装しています。 消費者向け、管理用、トランザクション用APIは利用できません。 APIは書き込み専用です。

Kafka互換APIの使用時期

Kafka互換APIは以下のシナリオに最適です:

• Zerobus SDKを採用しずにDeltaにデータを送信したい場合、すでにKafkaプロデューサーやアプリケーション、エージェント、コレクタを運用してKafkaに送信しています。
• 既存のKafkaプロデューサーの設定、バッチ処理、運用ツールを再利用したいのです。
• JSONレコードを送信すれば、プロトコルバッファ(protobuf)やApache Arrowは必要ありません。

新しいクライアントを一から構築し、最高スループット、レコードごとの確認応答、自動復旧を求める場合は、Kafka互換APIではなくZerobus SDKをgRPCで使用してください。 「 インターフェイスの選択」を参照してください。 列型またはバッチ型のワークロードについては、Arrow Flight と Zerobus Ingest の併用を参照してください。

インジェストモデルの仕組み

Kafka互換APIは、Kafkaの概念をZerobus Ingestに次のようにマッピングします:

• トピック。 Kafkaのトピック名は、Unity Catalogの3レベルテーブル名(catalog.schema.table)の完全な名称です。 ターゲットテーブルはすでに存在しているはずです。なぜならZerobusはトピックを決して作成しないからです。
• レコード。 Zerobusはレコード値のみを取り込みますが、それはターゲットのDeltaテーブルスキーマに一致するUTF-8エンコードされたJSONオブジェクトでなければなりません。 レコードキー、ヘッダー、クライアント提供のパーティションやタイムスタンプを無視し、永続化しません。
• 認証。 各接続は、ターゲットテーブルにスコープされたAzure Databricks OAuthトークンで認証され、SASL/OAUTHBEARERを通じて提示されます。 認証に関するページを参照してください。
• 謝辞。 ゼロバスは、レコードを持続的に永続化した後にのみ Produce 応答を返します。 プロデューサーを acks=allで設定してください。

ゼロバスは設計上パーティションレスです。 Zerobusエンドポイントは単一の論理ブローカー兼パーティションであり、トピックのすべてのレコードはパーティション 0上で認識されます。 プロデューサーはこれを考慮する必要はありません。 Zerobus Ingestは、入荷負荷に対応するために水平方向にスケールします。

また、Zerobus Ingestは少なくとも一度の配送を提供します。 Kafka互換APIはベータ版で独自のクォータがあり、Zerobus Ingestの他のAPIと同様の遅延、レコードサイズ、パーティションテーブル特性を共有しています。 ベータ割当やその他の割当については、 ゼロバスの取り込み割当についてご参照ください。

認証

Kafka互換APIはSASL/OAUTHBEARERを使用しています。 ベアラートークンは、ターゲットテーブルにアクセスできるサービスプリンシパルのクライアント認証情報で取得できるAzure Databricks OAuthトークンです。 トークンはOAuth authorization_detailsを通じてそのテーブルにスコープが割り当てられ、Zerobus REST APIと同じフローであるzerobusDirectWriteApiリソースを使用します。

OAuthトークンは1時間で期限切れになるため、静的文字列ではなくKafkaクライアントのトークンプロバイダーコールバックを通じてトークンを供給してください。 クライアントは再接続するたびに新しいトークンを再取得します。 接続にはサーバー側の寿命が制限されています。 接続がその限界に達すると、Zerobusはそれをクローズし、プロデューサーは自ら再接続して再認証を行います。 ライブ接続での再認証はサポートされていません。

接続前にターゲットテーブルで必要なUnity Catalog権限をサービスプリンシパルに付与してください。 「 サービスプリンシパルを作成し権限を付与する」を参照してください。

クライアントを書き込む

以下の例は、Use Zerobus Ingestの例で使われたのと同じair_qualityテーブルを生成します。 kafka-pythonを使用していますが、OAUTHBEARERメカニズムでSASL_SSLをサポートするKafkaプロデューサークライアントならどれでも動作します。 トークンプロバイダーのパターンをクライアントライブラリに合わせて適応させてください。

プロデューサーはポート 9092でZerobusブートストラップサーバーに接続します。 「 Get your workspace URL」と「Zerobus Ingest」エンドポイントで説明されているように、ワークスペースIDと地域を見つけてください。

• Bootstrapサーバー:<workspace-id>.zerobus.<region>.azuredatabricks.net:9092

pip install kafka-python requests

ステップ1:トークンプロバイダーを構築する

Zerobus は、有効期間の短いテーブル単位の Azure Databricks OAuth トークンを使用して、各接続を認証します。 トークンが期限切れになるため、静的トークンではなく、オンデマンドで新しいトークンを発行するコールバックを渡してください。

fetch_zerobus_token() 関数は、サービス プリンシパルの資格情報を対象テーブルをスコープとするトークンに交換し、ZerobusTokenProvider そのトークンを kafka-python が想定するコールバック インターフェイスでラップします。

import json

import requests
from kafka.sasl.oauth import AbstractTokenProvider

# See "Get your workspace URL and Zerobus Ingest endpoint" in zerobus-ingest.md.
WORKSPACE_ID = "1234567890123456"
WORKSPACE_URL = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
TABLE_NAME = "main.default.air_quality"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"


def fetch_zerobus_token():
    catalog, schema, table = TABLE_NAME.split(".")
    authorization_details = [
        {
            "type": "unity_catalog_privileges",
            "privileges": ["USE CATALOG"],
            "object_type": "CATALOG",
            "object_full_path": catalog,
        },
        {
            "type": "unity_catalog_privileges",
            "privileges": ["USE SCHEMA"],
            "object_type": "SCHEMA",
            "object_full_path": f"{catalog}.{schema}",
        },
        {
            "type": "unity_catalog_privileges",
            "privileges": ["SELECT", "MODIFY"],
            "object_type": "TABLE",
            "object_full_path": TABLE_NAME,
        },
    ]

    response = requests.post(
        f"{WORKSPACE_URL}/oidc/v1/token",
        auth=(CLIENT_ID, CLIENT_SECRET),
        data={
            "grant_type": "client_credentials",
            "scope": "all-apis",
            "resource": f"api://databricks/workspaces/{WORKSPACE_ID}/zerobusDirectWriteApi",
            "authorization_details": json.dumps(authorization_details),
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["access_token"]


# kafka-python calls token() whenever it needs a fresh OAuth token.
class ZerobusTokenProvider(AbstractTokenProvider):
    def token(self):
        return fetch_zerobus_token()

ステップ2:プロデューサーを設定し、レコードを送信する

プロデューサーをブートストラップサーバーに向け、OAUTHBEARERメカニズムでSASL_SSLを設定し、ステップ1のトークンプロバイダーを渡します。 各バッチが永続化された後にのみ認識される acks="all" を使い、レコードを圧縮せずに送信してください。

from kafka import KafkaProducer

BOOTSTRAP_SERVERS = "1234567890123456.zerobus.us-west-2.cloud.databricks.com:9092"

producer = KafkaProducer(
    bootstrap_servers=BOOTSTRAP_SERVERS,
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    sasl_oauth_token_provider=ZerobusTokenProvider(),
    # Wait for durable acknowledgment before treating a record as ingested.
    acks="all",
    # Compression is not supported by the endpoint; send records uncompressed.
    compression_type=None,
)

# Each send() returns a future immediately. The topic is the full table name.
futures = [
    producer.send(
        topic=TABLE_NAME,
        value=json.dumps(
            {"device_name": f"sensor-{i}", "temp": 20 + i % 15, "humidity": 50 + i % 40}
        ).encode("utf-8"),
    )
    for i in range(1000)
]

producer.flush()

# Block on each future to confirm every record was durably acknowledged.
for future in futures:
    future.get(timeout=30)

producer.close()
print("All records ingested successfully")

構成オプション

Kafka互換APIはKafkaプロトコルのプロデューサーサブセットを実装しています。 以下のオプションでプロデューサーを設定してください。

遅延、クォータ、レコードサイズ、パーティションテーブルの制限については、 Zerobus Ingest quotaを参照してください。

ベスト プラクティス

Kafka互換APIから最高のパフォーマンスと信頼性を得るために、以下のガイドラインに従ってください。

• バッチごとにプロデューサーを作成するのではなく、長期間使用できるプロデューサーを多くのレコードにまたがって再利用してください。プロデューサーの作成や SASL ハンドシェイクには初期化コストがかかります。
• プロデューサーがレコードをバッチにまとめて蓄積できるようにしましょう。例えば、レコードごとにフラッシュするのではなく、 linger.ms や、 batch.sizeをチューニングするなどです。 バッチ処理はスループットを最も大きく左右する手段です。
• acks=all を使用すると、Zerobus の「少なくとも 1 回」というセマンティクスに準拠し、各バッチに対して永続的な確認応答を取得できます。
• クライアントのトークンプロバイダーコールバックを通じてOAuthトークンを取得し、再接続時に自動的にリフレッシュできるようにしてください。期限切れの静的トークンを渡すのではなく。
• 最大のスループットを得るために、Zerobusエンドポイントと同じクラウドリージョン内でプロデューサーを稼働させましょう。

エラー処理

Zerobusは、影響を受けたトピックとパーティションで標準的なKafkaエラーコードを使って障害を報告します。 一般的なコードには、次のようなものがあります。

Produceリクエストが失敗すると、ゼロバスはパーティションごとのエラーコードを返し、接続を終了します。 Kafkaプロデューサーは自動的に再接続しますが、クライアントは送信失敗を表面化するよう設計します。例えば、各送信の結果を検査することで、レコードが静かにドロップしないようにします。

その他のリソース

• Zerobus Ingestを活用してください:まだZerobus Ingestを設定していない場合は、ワークスペースURLの見つけ方、ターゲットのDeltaテーブルの作成、サービスプリンシパルの設定手順をこちらからご覧ください。 これらのステップはすべてのインターフェースで共有されます。
• Zerobus Ingest クォータ:本番環境にデプロイする前にデフォルトの Zerobus クォータを確認しましょう。
• Arrow Flight と Zerobus Ingest を使用: gRPC による列単位またはバッチ単位のインジェストに使用されます。