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

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 にレコードをプッシュすると、その 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 データの取り込みを参照してください。
  • Kafka 互換 APIs (Beta): Databricks SDK を使用せず、既存の Apache Kafka プロデューサーを Zerobus Ingest に向けます。Zerobus Ingest での Kafka 互換 APIs の使用を参照してください。

Zerobus Ingest のスケーリングアーキテクチャ：ソースは Protocol Buffers (protobuf)、 JSON、および Arrow レコードを gRPC、 REST、 OpenTelemetry、および Kafka 互換の APIs を介して送信します。これらはオートスケーリングとロードバランシングを経て、ステートレスな Zerobus ノードの水平方向にスケーラブルなプールに流れます。各ノードには先行書き込みログと Lakehouse ライターがあり、レコードを Unity Catalog 管理下の Delta テーブルにバッチ commit します。

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

その他のリソース​

このページの見出し