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

Zerobus Ingest の概要

Zerobus Ingest は、実行すべきメッセージバスを必要とせず、大規模なデータを Unity Catalog Delta テーブルに直接書き込むプッシュベースのストリーミング API です。これにより、多くのチームがプロデューサーとレイクハウスの間に配置している中間レイヤーが削除されます。ワークフローは、 テーブルを作成し、そこにデータをプッシュする という2つのステップで構成されます。「hello world」クライアントとペタバイト規模のワークロードは、管理すべきインフラストラクチャなしで、本質的に同じコードを実行します。

メッセージバスを使用した取り込みでは、Deltaテーブルに到達する前にブローカーと取り込みジョブを経由してプロデューサーがルーティングされますが、Zerobus Ingestはプロデューサーをlakehouseに直接接続します。

Zerobus Ingest は Serverless であり、負荷の変化に応じてキャパシティを追加および削除します。24時間未満で1兆件を超えるレコードを単一のテーブルに取り込み（Ingesting the Milky Way: Petabyte-Scale with Zerobus Ingest ブログ記事を参照）、数秒以内にレコードを格納します。

  • サポートされているリージョンの一覧については、「 取り込みの可用性」を参照してください。

利点​

Zerobus Ingestは、最大規模のワークロードにスケーリングしながら、取り込みをシンプルに保ちます：

  • 設計によるシンプルさ。 テーブルを作成し、そこにデータをプッシュします。管理すべきブローカー、パーティション、パイプラインはありません。データが格納される前にメッセージバスやインジェクションジョブを経由させるのではなく、プロデューサーが直接テーブルに書き込むため、ホップ数や操作が必要な可動パーツが少なくなります。
  • Serverless でエラスティック。 Zerobus Ingest は default でオンになっており、負荷の変化に応じて容量を追加または削除します。アプリケーションを書き換えるのではなく、プロデューサーを増やすことでスケールします。その方法については、「Zerobus Ingest のスケーリング方法」を参照してください。
  • High-throughput workloads. Zerobus Ingestは大規模な取り込み用に構築されており、単一テーブルへの高い書き込みレートを維持します。
  • ほぼリアルタイムの鮮度。 レコードは数秒以内に Delta に格納され、到着後すぐにクエリーを実行できる状態になります。
  • high concurrency. Zerobus Ingest は、数千のクライアントから同一テーブルへの並列書き込みを処理します。

宛先がlakehouseの場合、Zerobus Ingestが最も直接的なパスとなります。その他のDatabricksツールは、隣接するニーズに対応し、これと併用することで効果を発揮します：

  • レイクハウス以外のコンシューマーをサポートするためにKafkaを実行しているユースケースでは、レイクハウス内にデータのコピーを作成することも検討してください。マネージドストリーミングコネクタを使用してレプリケートします。
  • クラウドストレージにファイルとして既に保存されているデータについては、Auto Loader を使用してください。
  • 処理パスで1秒未満の運用レイテンシが必要な場合は、リアルタイムモードを使用してください。

テーブルを作成し、データをプッシュする​

Zerobus Ingestの使用は、テーブルを作成してデータをプッシュするのと同じくらい簡単です。テーブルのスキーマは、各レコードに何を含める必要があるかを定義します。まず、ターゲットテーブルを作成します：

SQL
CREATE TABLE main.default.air_quality (
  device_name STRING,
  temp INT,
  humidity INT
);

次に、レコードの取り込みは数行のコードで行えます。

Python
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})
# ingest more records...
stream.close()

開発環境で出荷するコードと同じものを、本番運用のワークロードにスケールさせることができます。詳細なウォークスルーについては、Zerobus Ingest の使用を参照してください。

一般的なユースケース​

  • IoT およびデバイステレメトリ : センサー、車両、スマートデバイスのデータを、大規模な分散フリートから直接、ガバナンスされた Delta テーブルにストリームします。
  • オンプレミスからクラウドへ : 中間にブローカーインフラストラクチャを構築することなく、オンプレミスおよびハイブリッドシステムをレイクハウスに接続します。プライベート接続とファイアウォールの設定については、ネットワークに関する考慮事項を参照してください。
  • アプリケーションおよびクリックストリームイベント ：ニアリアルタイムのアナリティクス用に、クラウドおよびエッジアプリケーションからイベントをプッシュします。
  • チェンジデータキャプチャ (CDC) : 運用システムからの行の変更を Delta に取り込みます。
  • 可観測性データ : OpenTelemetry トレース、Logs、メトリクスを所有する Delta テーブルに送信します。Zerobus Ingest を使用した OpenTelemetry データの取り込みを参照してください。

仕組み​

プロデューサーは Zerobus Ingest に接続し、レコードをターゲット Delta テーブルに送信します。サービスは、テーブルスキーマに対して各レコードを検証し、受け入れられたレコードを永続化します。耐久性の確認を提供するインターフェイスの場合、Zerobus Ingest はレコードが永続化した後に確認応答を返します。データはその後まもなく（通常は数秒以内に）テーブルにマテリアライズされます。Zerobus Ingest の動的でパーティションレスな設計により、取り込みが弾力的に行われるため、Serverless コンピュートはワークロードに合わせてスケーリングします。

Zerobus Ingest の仕組み：プロデューサーがレコードを Zerobus Ingest Endpoint に送信すると、検証、受け入れられたレコードの永続化が行われ、インターフェイスから永続性の確認応答が提供された場合に確認応答が返され、Unity Catalog Delta テーブルにマテライゼーションされます

ストリームの詳細や Zerobus Ingest のスケーリング方法については、Zerobus Ingest の概念を参照してください。非同期クライアント/サーバー通信モデルについては、非同期通信を参照してください。

データを送信する方法​

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 データの取り込みを参照してください。
  • MQTT : TLS経由で既存の MQTT v5 アプリケーションから JSON レコードをパブリッシュします。Zerobus Ingest で MQTT を使用するを参照してください。
  • Kafka 互換 APIs : Databricks SDK を使用せずに、既存の Apache Kafka プロデューサーを Zerobus Ingest に向けます。詳細については、Zerobus Ingest での Kafka 互換 APIs の使用を参照してください。

次の図は、ソースストリームおよびAPIからのレコードが、自動スケーリングとロードバランシングを介して、水平方向のスケーリングが可能なステートレスなZerobusノードのプールへとどのように流れるかを示しています。各ノードには、先行書き込みログ（write-ahead log）とレイクハウスライターがあり、Unity Catalogが管理するDeltaテーブルにレコードをバッチcommitします。図内のソースラベルは代表的なものです。MQTTは、ロードバランシング、耐久性のあるストレージ、およびテーブルのマテリアライゼーションを通じて、同じパスに従います。

ソースAPIからDeltaテーブルにレコードをバッチcommitするステートレスノードを備えたZerobus Ingestのスケーリングアーキテクチャ図。

これらはすべて、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)、'MQTT' (MQTT)。

その他のリソース​

このページの見出し