Zerobus Ingest で Kafka 互換 APIs を使用する
ベータ版
この機能はベータ版であり、AWSおよびAzureでのみ利用可能です。
Zerobus Ingest は、Databricks SDK を使用せずに任意の Apache Kafka プロデューサークライアントを使用してインジェストできる、Kafka 互換のプロデューサー APIs を提供します。既存の Kafka プロデューサーを Zerobus Endpoint に向け、ターゲットテーブル名にちなんだトピックにプロデュースすると、レコードは Unity Catalog の Delta テーブルに格納されます。 [[ ## completed ##]]Kafka 互換の APIs は、すでに Kafka プロデューサー、Kafka プロトコルに対応したコレクター、または Kafka に出力するツールがあり、最小限のコード変更でそのデータを Delta にルーティングしたい場合に適しています。
Kafka 互換 APIs は、SASL_SSL および OAUTHBEARER メカニズムを使用して TCP 上でランし、Kafka プロトコルのプロデューサー側のサブセットを実装します。これには Produce、Metadata、ApiVersions、および SASL ハンドシェイク APIs が含まれます。コンシューマー、管理者、およびトランザクション APIs は利用できません。APIs は書き込み専用です。
Kafka互換APIを使用する場合
Kafka互換APIは、以下のシナリオに最適です。
- Zerobus SDK を採用せずに Delta にデータを送信したい場合で、すでに Kafka プロデューサー、または Kafka にデータを送信するアプリケーション、エージェント、コレクターをランしている場合。
- 既存の Kafka プロデューサー構成、バッチ処理、および運用ツールを再利用したい場合。
- JSON レコードを送信しており、Protocol Buffers や Apache Arrow を必要としない場合。
ゼロから新しいクライアントを構築し、最高のthroughput、レコードごとの確認応答、および自動回復を求める場合は、Kafka互換APIの代わりにgRPC経由でZerobus SDKを使用してください。「インターフェースの選択」を参照してください。列指向またはバッチ処理のワークロードについては、 「Zerobus Ingest での Arrow Flight の使用」を参照してください。
取り込みモデルの仕組み
Kafka互換のAPIsは、Kafkaの概念を次のようにZerobus Ingestにマッピングします。
- トピック。 Kafka トピック名は、完全な 3 レベルの Unity Catalog テーブル名 (
catalog.schema.table) です。Zerobus はトピックを作成しないため、ターゲットテーブルは既に存在している必要があります。 - レコード。 Zerobusはレコード値のみを取り込みます。これは、ターゲットのDeltaテーブルスキーマと一致するUTF-8でエンコードされたJSONオブジェクトである必要があります。これはレコードキー、ヘッダー、およびクライアントが提供したパーティションとTimestampを無視し、それらを永続化しません。
- 認証。 各接続は、ターゲットテーブルに対してスコープ設定された Databricks OAuth トークンを使用して認証され、SASL/
OAUTHBEARERを介して提示されます。認証を参照してください。 - 了承事項。 Zerobus は、レコードを永続的に保存した後にのみ
Produceレスポンスを返します。acks=allを使用してプロデューサーを構成します。
Zerobus は設計上、パーティションレスです。Zerobus Endpointは単一の論理ブローカーおよびパーティションであるため、トピックのすべてのレコードはパーティション 0 で確認応答されます。プロデューサー側でこれをアカウントする必要はありません。Zerobus Ingest は、受信負荷を処理するために水平方向にスケールします。
また、Zerobus Ingest は少なくとも1回の配信（at-least-once delivery）を提供します。単一の接続で1秒あたり約50,000メッセージを維持できますが、これは SDK gRPC パスよりも低い値です。最高の throughput を得るには、Kafka 互換の APIs ではなく Zerobus SDK を使用してください。レイテンシ、クォータ、レコードサイズ、およびパーティション分割されたテーブルの制限は、Zerobus Ingest の他の機能と共有されます。Zerobus Ingest コネクタのクォータを参照してください。
認証
Kafka 互換 APIs は SASL/OAUTHBEARER を使用します。ベアラー トークンは、ターゲットテーブルへのアクセス権を持つ Service Principalのクライアント資格情報を使用して取得する Databricks OAuth トークンです。トークンは OAuth authorization_details を通じてそのテーブルにスコープされ、zerobusDirectWriteApi リソースを使用します。これは Zerobus REST API と同じフローです。
OAuth トークンの有効期限は 1 時間であるため、静的な文字列としてではなく、Kafka クライアントのトークンプロバイダー コールバックを通じてトークンを提供してください。その後、クライアントは再接続するたびに新しいトークンを再取得します。接続には、サーバー側で制限された有効期間もあります。接続がその制限に達すると、Zerobus は接続を閉じ、プロデューサーは自動的に再接続および再認証を行います。アクティブな接続での再認証はサポートされていません。
接続する前に、ターゲットテーブルに対して必要なUnity Catalog権限をService Principalに付与してください。「Service Principalの作成と権限の付与」を参照してください。
クライアントを記述する
以下の例では、Zerobus Ingest コネクタの使用の例で使用されているものと同じ air_quality テーブルに対して生成を行います。これは kafka-python を使用しますが、OAUTHBEARER メカニズムで SASL_SSL をサポートする Kafka プロデューサークライアントであれば動作します。トークンプロバイダーパターンをクライアントライブラリに適応させてください。
プロデューサーは、ポート 9092 でZerobusブートストラップサーバーに接続します。「ワークスペースURLとZerobus Ingest Endpointの取得」で説明されているように、ワークスペースIDとリージョンを確認します。
[[ ## completed ##]]
- ブートストラップサーバー:
<workspace-id>.zerobus.<region>.cloud.databricks.com:9092
pip install kafka-python requests
ステップ1: トークンプロバイダーを構築する
Zerobus は、有効期間が短く、テーブルスコープの Databricks OAuth トークンを使用して各接続を認証します。トークンには有効期限があるため、静的なトークンではなく、必要に応じて新しいトークンを発行するコールバックを渡してください。
fetch_zerobus_token() 関数は、Service Principal の資格情報をターゲットテーブルにスコープされたトークンと交換し、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: プロデューサーを構成してレコードを送信する
プロデューサーをブートストラップサーバーに向け、SASL_SSL を OAUTHBEARER メカニズムで構成し、ステップ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 acknowledgement 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 互換の APIs は、Kafka プロトコルのプロデューサーサブセットを実装しています。次のオプションに従ってプロデューサーを構成します。
オプション | 詳細 |
|---|---|
レコード形式 | JSONのみ。各レコードの値は、ターゲット テーブルのスキーマと一致するUTF-8でエンコードされたJSONオブジェクトである必要があります。Protocol BuffersまたはAvroを送信するには、Zerobus SDKを使用します。 |
圧縮 | サポートされていません。圧縮されていないバッチを送信してください（例: |
レコードフィールド | 値のみ。Zerobusはレコード値を取り込みますが、キー、ヘッダー、パーティション割り当て、またはTimestampは保持しません。 |
API サポート | 書き込み専用。Zerobus は、 |
フロントエンドPrivateLink | サポートされていません。代わりにパブリックEndpoint経由で接続します。 |
スキーマ | 強制されます。Zerobus はターゲットテーブルのスキーマと一致しないフィールドを持つレコードを拒否し、追加の null 許容 Delta 列は破壊的でない変更として扱います。一致しないフィールドを拒否せずにキャプチャするには、レスキュー列を構成します。 |
フロントエンドPrivate Linkの情報については、「Private Link の概念」を参照してください。
レイテンシ、クォータ、レコードサイズ、およびパーティション分割テーブルの制限については、Zerobus Ingest コネクタのクォータを参照してください。
ベストプラクティス
Kafka互換APIから最高のパフォーマンスと信頼性を得るには、これらのガイドラインに従ってください。単一の接続で、1秒あたり約50,000メッセージを維持できます。
- プロデューサーの作成と SASL ハンドシェイクにはセットアップコストがかかるため、バッチごとに作成するのではなく、長期間存続するプロデューサーを多くのレコード間で再利用してください。
- レコードごとにフラッシュするのではなく、たとえば
linger.msやbatch.sizeを調整して、プロデューサーがレコードをバッチに蓄積するようにします。バッチ処理は、throughputを向上させる最大の手段です。 - 各バッチの永続的な確認応答を取得するには
acks=allを使用します。これは Zerobus の「最低 1 回」のセマンティクスと一致します。 - 有効期限が切れる静的なトークンを渡すのではなく、クライアントのトークンプロバイダーコールバックを通じてOAuthトークンを取得し、再接続時に自動的に更新されるようにします。
- 最大のthroughputを得るには、Zerobus Endpointと同じクラウドリージョンでランしてください。
エラー処理
Zerobus は、影響を受けるトピックとパーティションで標準の Kafka エラーコードを使用して障害を報告します。一般的なコードは次のとおりです。
Kafka エラー | 意味 |
|---|---|
| OAuth トークンが存在しないか、無効であるか、またはテーブルに対する必要な Unity Catalog 権限が不足しています。 |
| ターゲットテーブルが存在しないか、削除されたか、またはトークンに書き込み権限がありません。 |
| レコードがスキーマ検証に失敗したか、UTF-8 JSONとしてデコードできませんでした。 |
| 単一のレコードが 10 MB のレコードサイズ制限を超えています。レコードサイズを参照してください。 |
| バッチは圧縮されました。レコードを非圧縮で送信します。 |
Produce リクエストが失敗した後、Zerobus はパーティションごとのエラーコードを返し、接続を閉じます。Kafka プロデューサーは自動的に再接続しますが、各送信の結果を検査するなどして送信失敗が表面化するようにクライアントを設計し、レコードが警告なしに破棄されないようにしてください。
その他のリソース
- Zerobus Ingest コネクタの使用: Zerobus Ingest をまだ設定していない場合は、ここから起動して、ワークスペース URL の検索、ターゲット Delta テーブルの作成、および Service Principal の構成に関する手順を確認してください。 [[ ## completed ##]]これらのステップは、すべてのインターフェースで共通です。
- Zerobus Ingest コネクタのクォータ: 本番運用にデプロイする前に、Zerobus のクォータと制限を確認してください。
- Zerobus Ingest で Arrow Flight を使用する: gRPC を介したカラムナまたはバッチ取り込み用。