Zerobus Ingest の概要
Zerobus Ingest は、実行すべきメッセージバスを必要とせず、大規模なデータを Unity Catalog Delta テーブルに直接書き込むプッシュベースのストリーミング API です。ワークフローは、 テーブルを作成し、そこにデータをプッシュする という2つのステップで構成されます。Zerobus Ingest を使用すると、パーティション、ブローカー、またはパイプラインを管理する必要がなくなります。これはワークスペースでdefaultでオンになっており、接続を開くにつれてスケールするServerless Endpointです。
高スループットとリアルタイムに近い鮮度を実現するために構築された Zerobus Ingest は、数千のクライアントから同じテーブルへの大量の並列書き込みを処理し、数秒以内にレコードを Delta に格納するため、データが到着してすぐにクエリーを実行できる状態になります。
- Zerobus Ingest は、特定のリージョンで利用可能です。ワークスペースとターゲットテーブルの両方が、サポートされているリージョン内にある必要があります。サポートされているリージョンの一覧については、「 Ingestion availability」を参照してください。
高いスケーラビリティを実現するように構築
Zerobus Ingest は、ユーザーが容量を計画することなく、高いスケーラビリティを実現するように設計されています。「Ingesting the Milky Way: Petabyte-Scale with Zerobus Ingest」ブログ記事で説明されているように、24時間未満で1兆件を超えるレコードを単一のDeltaテーブルに取り込み、数千のクライアントからの大容量の並列書き込みを処理しました。For default throughputクォータについては、「Zerobus Ingestクォータ」を参照してください。
「hello world」クライアントとペタバイト規模のワークロードは、本質的に同じコードを実行します。 アプリケーションを書き直すのではなく、プロデューサーを増やすことでスケールします。
Zerobus Ingest は Serverless です。負荷の変化に応じて容量を追加および削除します。ストリームは動的なパーティション単位として機能し、サービスは需要の変化に応じて容量を再調整するために、それらを開閉およびローテーションします。
Zerobus Ingestがこれをどのように実現するかについては、How Zerobus Ingest scalesを参照してください。
メッセージバスは不要です
多くのチームは、データを lakehouse に取り込む過程でバッファリングする目的のみで、プロデューサーとテーブルの間に Kafka などのメッセージバスを配置しています。それにより、ホップ数、コスト、運用オーバーヘッドが増加します。具体的には、ブローカーのサイジング、パーティションの再バランス、コンシューマーラグの監視が必要になります。Zerobus Ingest はその中間層を排除し、プロデューサーが Delta に直接書き込めるようにします。
同じデータがレイクハウス以外の多くのコンシューマーに供給される場合、マイクロサービス間のメッセージングが必要な場合、またはメッセージのファンアウトを行う場合には、依然としてメッセージバスが適切なツールです。そのような場合で、そのデータをレイクハウスにも保存したいときは、Databricks の マネージド ストリーミング コネクタ を使用してメッセージバスからレプリケートします。しかし、レイクハウスが宛先である場合、Zerobus Ingest の方がよりシンプルで直接的なパスとなります。
仕組み
プロデューサーは Zerobus Ingest へのストリームを開き、ターゲットの Delta テーブルにレコードをプッシュします。サービスは各レコードをテーブルスキーマと照らし合わせて検証し、永続化します。レコードが永続化されると、Zerobus Ingest は迅速に承諾を返すため、プロデューサーは各レコードの完了を待たずに送信を継続できます。データはその後まもなく、通常は数秒以内にテーブルにマテリアライズされます。Zerobus Ingest の動的でパーティションレスな設計により、取り込みが柔軟になり、Serverless コンピュートがワークロードに合わせてスケールします。
ストリームの詳細や Zerobus Ingest のスケーリング方法については、Zerobus Ingest の概念を参照してください。非同期クライアント/サーバー通信モデルについては、非同期通信を参照してください。
テーブルを作成し、データをプッシュする
Zerobus Ingest SDK を使用できる、またはサポートされている API (gRPC、REST、または OpenTelemetry) を呼び出せるアプリケーションであれば、Delta テーブルにデータをストリームできます。テーブルのスキーマは、各レコードに何を含める必要があるかを定義します。まず、ターゲットテーブルを作成します：
CREATE TABLE main.default.air_quality (
device_name STRING,
temp INT,
humidity INT
);
次に、Service Principal にテーブルへのアクセス権を付与した後、レコードの取り込みは数行のコードで行えます：
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties
sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)
table_properties = TableProperties("main.default.air_quality")
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)
stream.ingest_record_offset({"device_name": "sensor-1", "temp": 22, "humidity": 55})
stream.close()
1 つのレコードを取り込むのと同じコードでペタバイト規模までスケールします。より多くのプロデューサーから実行するだけです。詳細なウォークスルーについては、Zerobus Ingest の使用を参照してください。
Zerobus Ingest の使用時期
次のような場合には、Zerobus Ingest を使用します… | …の場合は、別のツールの使用を検討してください。 |
|---|---|
レイクハウスは、データの唯一の保存先です。 | 同じデータをレイクハウス以外の多くのコンシューマーにファンアウトする必要がある場合は、Kafkaなどのメッセージバスを使用し、ストリーミング コネクタを使用してそのデータをレイクハウスにレプリケートします。 |
Deltaテーブルへの高throughputな並列書き込みを直接行う場合。 | マイクロサービス間のメッセージングが必要です (メッセージバスを使用してください)。 |
ニアリアルタイムの鮮度（数秒）がニーズを満たします。 | 処理パスでサブ秒単位の運用レイテンシが必要な場合は、リアルタイムモードのコンセプトを使用してください。 |
プロデューサーを制御し、API にデータをプッシュできます。 | クラウドストレージに既に到着しているファイルからプルしている場合（Auto Loader を使用してください）。 |
計画すべき設計上のポイント：Zerobus Ingest は、ストリーム全体ではなく、ストリーム単位での順序を保証します。ストリーム単位の順序付けの仕組みと、それに基づいた設計方法については、ストリームを参照してください。
一般的なユースケース
- IoT およびデバイステレメトリ : センサー、車両、スマートデバイスのデータを、大規模な分散フリートから直接、ガバナンスされた Delta テーブルにストリームします。
- オンプレミスからクラウドへ : 中間にブローカーインフラストラクチャを構築することなく、オンプレミスおよびハイブリッドシステムをレイクハウスに接続します。プライベート接続とファイアウォールの設定については、ネットワークに関する考慮事項を参照してください。
- アプリケーションおよびクリックストリームイベント ：ニアリアルタイムのアナリティクス用に、クラウドおよびエッジアプリケーションからイベントをプッシュします。
- チェンジデータキャプチャ (CDC) : 運用システムからの行の変更を Delta に取り込みます。
- 可観測性データ : OpenTelemetry トレース、Logs、メトリクスを所有する Delta テーブルに送信します。Zerobus Ingest を使用した OpenTelemetry データの取り込みを参照してください。
データを送信する方法
Zerobus Ingest は複数のインターフェースをサポートする 1 つの Endpoint であるため、各プロデューサーに最適なものを選択できます：
- gRPC 経由の SDK : Python、Java、Rust、Go、TypeScript、および（ベータ版の）C++ と C# / .NET での high-throughput なストリーミングクライアント。大量の順序付けられたインジェストに最適です。「 Write a client」を参照してください。
- REST API : エッジデバイスの大規模なフリートなどの軽量または「チャッティ（頻繁に通信する）」なクライアント向けのステートレスインターフェイス。クライアントを記述するを参照してください。
- OpenTelemetry (OTLP) : 既存の OpenTelemetry コレクターを Zerobus Ingest に向けることで、カスタム統合なしでトレース、Logs、メトリクスを取り込めます。Zerobus Ingest を使用した OpenTelemetry データの取り込みを参照してください。
- Kafka 互換 APIs (Beta): Databricks SDK を使用せず、既存の Apache Kafka プロデューサーを Zerobus Ingest に向けます。Zerobus Ingest での Kafka 互換 APIs の使用を参照してください。
これらはすべて、Delta テーブルに直接書き込みます。完全な比較および選択方法については、API プロトコルを参照してください。最初のクライアントを作成するには、Zerobus Ingest の使用を参照してください。
料金
Zerobus Ingest の料金は、「Jobs Serverless」SKU に対して請求されます。料金については、Lakeflow Connect 価格ページをご覧ください。
使用状況をモニタリングする
課金利用システムテーブルを通じて、利用料金を監視できます。「課金利用システムテーブルリファレンス」を参照してください。以下を使用して Zerobus Ingest の利用状況をフィルタリングします：
billing_origin_product = 'LAKEFLOW_CONNECT'product_features.lakeflow_connect.zerobus_request_typeデータの取り込み方法を識別します:'GRPC'(SDK ストリーミング)、'HTTP'(REST)、'OTEL_GRPC'および'OTEL_HTTP'(OpenTelemetry/OTLP)、または'KAFKA'(Kafka 互換 APIs)。