Zerobus Ingest を使用する

このページでは、Lakeflow Connectで Zerobus Ingest を使用してデータを取り込む方法について説明します。

Zerobus Ingest を開始する

注記

クライアント側のファイアウォールを使用している場合は、Zerobus Ingestで使用されるIPアドレスを許可リストに追加してください。リージョン別のIPアドレスを表示するには、「Databricksのサービスと資産のIPアドレスとドメイン」を参照してください。

開始する前に、ワークスペースのリージョンで Zerobus Ingest が利用可能であることを確認してください。取り込みの可用性を参照してください。

1. Zerobus Ingest URL を取得します。
2. データを取り込むテーブルを作成するか、既存のテーブルを特定します。
3. Service Principalを作成し、テーブルへの権限を付与します。
4. クライアントまたはエクスポーターを接続して、データの送信を開始します。

ユースケースのガイドを選択してください：

• 独自のデータのインジェスト : 定義したスキーマで Zerobus Ingest SDK または REST API を使用します。このページの手順に従ってください。

• OpenTelemetry データの取り込み : 標準の OpenTelemetry SDK またはコレクターを使用して、トレース、Logs、メトリクスを定義済みのテーブルスキーマに送信します。詳細な手順については、 Zerobus Ingest を使用した OpenTelemetry データの取り込みを参照してください。

インターフェースを選択

Zerobus Ingest は複数のインターフェースをサポートしており、すべて Unity Catalog の Delta テーブルに直接書き込みます。要約：

• SDKs over gRPC ：最も高い持続的throughputを実現し、大容量のストリーミングプロデューサーに最適です。

• REST : ステートレスであり、軽量または「頻繁に通信する」エッジデバイスの大規模なフリートに最適です。

• OpenTelemetry (OTLP) : OpenTelemetryのトレース、Logs、メトリクスをすでに送信しているシステム用。Zerobus Ingestを使用したOpenTelemetryデータの取り込みを参照してください。

• Kafka 互換 APIs (ベータ版): すでに Kafka プロトコルに対応しているプロデューサー向け。Zerobus Ingest での Kafka 互換 APIs の使用を参照してください。

完全な比較および選択方法については、API プロトコルを参照してください。SDK を介して、レコード形式（JSON、Protocol Buffers (protobuf)、または Apache Arrow）を選択することもできます。メッセージタイプを参照してください。このページの残りの部分では、SDK および REST API を使用します。

ワークスペース URL と Zerobus Ingest エンドポイントを取得する

ログインすると、ワークスペースの URL がブラウザに表示されます。完全な URL はhttps://<databricks-instance>.com/o=XXXXX形式に従いますが、ワークスペース URL は/o=XXXXXより前のすべての内容で構成されます。たとえば、次の完全な URL を指定すると、ワークスペースの URL とワークスペース ID を特定できます。

• 完全なURL: https://abcd-teste2-test-spcse2.cloud.databricks.com/?o=2281745829657864#
• ワークスペース URL: https://abcd-teste2-test-spcse2.cloud.databricks.com
• ワークスペースID: 2281745829657864

サーバー エンドポイントは、ワークスペースとリージョンによって異なります。

• サーバーエンドポイント: <workspace-id>.zerobus.<region>.cloud.databricks.com

ワークスペースのリージョンを確認するには、Databricks UI の上部ナビゲーションバーにあるワークスペーススイッチャーを開きます。リージョンは各ワークスペース名の下に表示されます（例：us-west-2）。また、アカウントコンソールの [ ワークスペース ] からも確認できます。

リージョンの可用性については、 Zerobus Ingest クォータを参照してください。

ターゲットテーブルを作成または識別する

データを取り込むターゲット テーブルを特定します。新しいターゲット テーブルを作成するには、 CREATE TABLE SQL コマンドを実行します。たとえば、 unity.default.air_qualityという名前の新しいテーブルを作成します。

SQL

    CREATE TABLE unity.default.air_quality (
    device_name STRING, temp INT, humidity LONG);

注記

OpenTelemetry 取り込みの場合、テーブルは信号タイプ (トレース、ログ、メトリクス) ごとに事前定義されたスキーマを使用する必要があります。 Unity Catalogでターゲットテーブルを作成するを参照してください。

テーブルスキーマは Zerobus Ingest が受け入れるものの契約であり、Zerobus Ingest が自動的に進化させることはありません。スキーマの変更を事前に計画します。最初にテーブルを進化させ、次にプロデューサーを更新してください。Zerobus Ingest は、破壊的なテーブル変更後に適合しなくなったレコードを、破棄するのではなく、永続的な fallback ロケーションに書き込みます。スキーマ管理と永続的な fallback ロケーションからのデータ回復を参照してください。

デフォルトでは、Zerobus Ingest はターゲットテーブルのスキーマと一致しないフィールドを持つレコードを拒否します。それらのフィールドを失わずにキャプチャするには、レスキュー列を構成します。Zerobus レスキュー列を参照してください。

ストリーミングテーブルへの取り込み

備考

ベータ版

Zerobus Ingest を使用して ストリーミングテーブル に取り込む機能は Beta 版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

新しいストリーミングテーブルを作成するには、CREATE STREAMING TABLEのSQLコマンドを実行します。例えば：

SQL

CREATE STREAMING TABLE unity.default.air_quality (
device_name STRING, temp INT, humidity LONG);

ストリーミングテーブルの作成後、標準の Delta テーブルの場合とまったく同様に、クライアントの書き込みにあるいずれかのインターフェースを使用してデータを取り込みます。ストリーミングテーブルへの書き込みは、管理された Delta テーブルへの書き込みと同じ方法で行われ、同じ制限とクォータが適用されます。

サービスプリンシパルを作成し、権限を付与する

A Service Principal は、パーソナライズされたアカウントよりも高いセキュリティを提供する特殊な ID です。Service Principalおよび認証での使用方法の詳細については、「OAuth を使用した Databricks へのService Principal アクセスの承認」を参照してください。

Databricks REST APIまたはSDKを使用してプログラムでService Principalを作成および管理するか、以下で説明するようにワークスペースUIを通じて行うことができます。このセクションの最後にある権限付与は、任意のクライアントから実行できるSQLです。

1. Service Principalを作成するには、 [設定] > [ID とアクセス] に移動します。

2. 「サービスシプリンパル」 で、 「管理」 を選択します。

3. [ サービスプリンシパルの追加 ] をクリックします。

4. [サービスプリンシパルの追加] ウィンドウで、 [新規追加 ] をクリックして新しいサービスプリンシパルを作成します。

5. サービスプリンシパルのクライアント ID とクライアント シークレットを生成して保存します。

6. カタログ、スキーマ、テーブルに必要な権限をサービスプリンシパルに付与します。

   1. Service Principal ページで、 Configurations tab に移動します。
   2. アプリケーション ID (UUID) をコピーします。
   3. 権限を付与するには、以下のSQLを使用してください。必要に応じて、例のUUID、カタログ名、スキーマ名、テーブル名を置き換えてください。
   SQL

   GRANT USE CATALOG ON CATALOG <catalog> TO `<UUID>`;
   GRANT USE SCHEMA ON SCHEMA <catalog.schema> TO `<UUID>`;
   GRANT MODIFY, SELECT ON TABLE <catalog.schema.table_name> TO `<UUID>`;

クライアントを記述する

ご希望のプログラミング言語で Zerobus SDK を使用するか、REST API を使用して、ターゲット テーブルにデータを取り込みます。SDK はオープンソースです。完全なライブラリ、言語固有のドキュメント、およびその他の例については、 Zerobus SDK リポジトリを参照してください。

以下の例では ingest_record_offset を使用しており、レコードを送信した順序が保持されます。

Python SDK

Python 3.9 以降が必要です。この SDK は、非同期ランタイムを通じて、高 throughput で効率的なネットワーク I/O を提供します。JSON（最もシンプル）および Protocol Buffers（本番運用に推奨）をサポートしています。この SDK は、同期および非同期の実装の両方と、オフセットベースおよびフューチャーベースの取り込み方法もサポートしています。

Bash

pip install databricks-zerobus-ingest-sdk

JSONの例:

Python

import logging
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties

# See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
SERVER_ENDPOINT="https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
DATABRICKS_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"

sdk = ZerobusSdk(
    SERVER_ENDPOINT,
    DATABRICKS_WORKSPACE_URL
)

table_properties = TableProperties(TABLE_NAME)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)

try:
    for i in range(1000):
        record_dict = {
            "device_name": f"sensor-{i}",
            "temp": 20 + i % 15,
            "humidity": 50 + i % 40
        }
        stream.ingest_record_offset(record_dict)
finally:
    stream.close()

上記の例では、返されたオフセットを待機せずに、オフセットベースの ingest_record_offset メソッドを使用しています。利用可能な取り込み方法、オフセットの耐久性確認を待機するタイミング、および確認コールバックを使用して進行状況を追跡する方法については、メッセージのブロッキングと確認を参照してください。

Protocol Buffers： 型安全な取り込みを行うには、protobuf 記述子を TableProperties に渡します（形式は自動的に選択されます）。generate_proto ツールを使用してテーブルからスキーマを生成し、protoc でコンパイルしてから、コンパイルされた記述子を渡してストリームを作成します。

Arrow Flight (ベータ版): 同じ gRPC 接続を介して Apache Arrow RecordBatchデータを列指向またはバッチ指向で取り込むには、 「Arrow Flight を Zerobus Ingest で使用する」を参照してください。[arrow]追加のpip install "databricks-zerobus-ingest-sdk[arrow]" pyarrowが必要です。

完全なドキュメント、構成オプション、バッチ取り込み、およびプロトコル バッファーの例については、 Python SDK リポジトリを参照してください。

Rust SDK

Rust 1.70以降が必要です。SDK は、高 throughput の取り込みのために非同期 I/O と gRPC を使用します。JSON（最もシンプル）と Protocol Buffers（本番運用に推奨）をサポートしています。

まず、パッケージをインポートします。

Bash

cargo add databricks-zerobus-ingest-sdk

または、 Cargo.tomlに追加します。

[dependencies]
databricks-zerobus-ingest-sdk = "2.0.0" # Latest version at time of publication

JSONの例:

Rust

  use databricks_zerobus_ingest_sdk::{JsonString, ZerobusSdk};
  use std::error::Error;

  // See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
  const DATABRICKS_WORKSPACE_URL: &str = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com";
  const SERVER_ENDPOINT: &str = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com";
  const TABLE_NAME: &str = "main.default.air_quality";
  const CLIENT_ID: &str = "your-client-id";
  const CLIENT_SECRET: &str = "your-client-secret";


  #[tokio::main]
  async fn main() -> Result<(), Box<dyn Error>> {
      let sdk_handle = ZerobusSdk::builder()
          .endpoint(SERVER_ENDPOINT)
          .unity_catalog_url(DATABRICKS_WORKSPACE_URL)
          .build()?;

      let mut stream = sdk_handle
          .stream_builder()
          .table(TABLE_NAME)
          .oauth(CLIENT_ID, CLIENT_SECRET)
          .json()
          .max_inflight_requests(100)
          .build()
          .await?;

      stream.ingest_record_offset(
        JsonString("{
          \"device_name\": \"sensor\",
          \"temp\": 22,
          \"humidity\": 55}".to_string())).await?;

      println!("Record ingested successfully");
      stream.close().await?;
      println!("Stream closed successfully");

      Ok(())
  }

Protocol Buffers: 型安全な取り込みを行うには、.json() の代わりにストリームビルダー上の .compiled_proto(descriptor) を介して Protocol Buffers を使用します。ここで、descriptor は prost_types::DescriptorProto です。generate_proto ツールを使用して必要なファイルを生成し、プロジェクトにインポートします。 Arrow Flight (Beta): 同じ gRPC 接続を介した Apache Arrow RecordBatch データの列指向またはバッチ指向の取り込みについては、「Use Arrow Flight with Zerobus Ingest」を参照してください。Cargo 機能 cargo add databricks-zerobus-ingest-sdk --features arrow-flight で有効化します。

完全なドキュメント、構成オプション、バッチ取り込み、 generate_protoツール、および Protocol Buffer の例については、 Rust SDK リポジトリを参照してください。

Java SDK

Java 8 以上が必要です。The SDK は、高 throughput の取り込みに対して低レイテンシで効率的なネットワーク I/O を提供します。JSON (最もシンプル) と Protocol Buffers (本番運用に推奨) をサポートしています。

Maven :

XML

<dependency>
    <groupId>com.databricks</groupId>
    <artifactId>zerobus-ingest-sdk</artifactId>
    <version>0.2.0</version>
</dependency>

JSONの例:

Java

import com.databricks.zerobus.*;

public class ZerobusClient {

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
    private static final String SERVER_ENDPOINT =
        "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com";
    private static final String DATABRICKS_WORKSPACE_URL =
        "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com";
    private static final String TABLE_NAME = "main.default.air_quality";
    private static final String CLIENT_ID = "your-client-id";
    private static final String CLIENT_SECRET = "your-client-secret";

    public static void main(String[] args) throws Exception {
        ZerobusSdk sdk = new ZerobusSdk(
            SERVER_ENDPOINT,
            DATABRICKS_WORKSPACE_URL
        );

        ZerobusJsonStream stream = sdk.streamBuilder()
            .table(TABLE_NAME)
            .oauth(CLIENT_ID, CLIENT_SECRET)
            .json()
            .build()
            .join();

        try {
            for (int i = 0; i < 100; i++) {
                String record = String.format(
                    "{\"device_name\": \"sensor-%d\", \"temp\": 22, \"humidity\": 55}", i
                );
                stream.ingestRecordOffset(record);
            }
        } finally {
            stream.close();
        }
    }
}

Protocol Buffers: 型安全な取り込みを行うには、streamBuilder() と .compiledProto(...) を使用して ZerobusProtoStream を作成します。バンドルされている JAR ツールを使用してテーブルからスキーマを生成し、protoc でコンパイルします。

Arrow Flight (ベータ版): 同じ gRPC 接続を介して Apache Arrow RecordBatchデータを列指向またはバッチ指向で取り込むには、 「Arrow Flight を Zerobus Ingest で使用する」を参照してください。

完全なドキュメント、構成オプション、バッチ取り込み、およびプロトコル バッファーの例については、 Java SDK リポジトリを参照してください。

Go SDK

Go 1.21 以降が必要です。SDK は、ストリーミング取り込みに対して高い throughput とパフォーマンスを提供します。JSON（最もシンプル）および Protocol Buffers（本番運用に推奨）をサポートしています。

Bash

go get github.com/databricks/zerobus-sdk/go@latest

JSONの例:

簡潔にするために、ここではエラーは無視されます。本番運用コードでは、常にエラーをチェックしてください。

Go

package main

import (
	"fmt"

	zerobus "github.com/databricks/zerobus-sdk/go"
)

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
const (
	ServerEndpoint         = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
	DatabricksWorkspaceURL = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
	TableName              = "main.default.air_quality"
	ClientID               = "your-client-id"
	ClientSecret           = "your-client-secret"
)

func main() {
	sdk, _ := zerobus.NewZerobusSdk(
		ServerEndpoint,
		DatabricksWorkspaceURL,
	)
	defer sdk.Free()

	options := zerobus.DefaultStreamConfigurationOptions()
	options.RecordType = zerobus.RecordTypeJson

	stream, _ := sdk.CreateStream(
		zerobus.TableProperties{
			TableName: TableName,
		},
		ClientID,
		ClientSecret,
		options,
	)
	defer stream.Close()

	_, _ = stream.IngestRecordOffset(`{
		"device_name": "sensor-001",
		"temp": 20,
		"humidity": 60
	}`)

  fmt.Println("Record ingested successfully")

  _ = stream.Close()
  fmt.Println("Stream closed successfully")
}

プロトコル バッファー: 型セーフな取り込みを行うには、 RecordTypeProto (デフォルト) でプロトコル バッファーを使用し、テーブル プロパティにdescriptorProtoを指定します。.protoを作成するテーブル スキーマに一致するファイルを検索し、 generate_protoスクリプトを実行してファイルをプロジェクトにインポートします。

Arrow Flight (ベータ版): 同じ gRPC 接続を介して Apache Arrow RecordBatchデータを列指向またはバッチ指向で取り込むには、 「Arrow Flight を Zerobus Ingest で使用する」を参照してください。

完全なドキュメント、構成オプション、バッチ取り込み、generate_proto ツール、およびプロトコル バッファーの例については、 Go SDK リポジトリを参照してください。

C++ SDK

備考

ベータ版

C++ SDKはベータ版です。

C++17 以上が必要です。この SDK は、RAII C++ インターフェースを通じて、ネイティブな gRPC ストリーミング、OAuth、および自動リカバリを提供します。シンプルなセットアップには JSON を、本番運用ワークロードには Protocol Buffers をサポートしています。

SDK はプラットフォームごとにビルド済みのリリースバンドルとして提供されるため、使用するために Rust ツールチェーンは必要ありません。リリースページからプラットフォーム（macOS、Linux（muslを含む）、またはWindows）用のバンドルをdownloadして展開し、CMake でバンドルされた FFI アーカイブを指定します。アーカイブ名は、macOS と Linux では libzerobus_ffi.a、Windows では zerobus_ffi.lib です：

Bash

# macOS and Linux
cmake -S cpp -B build \
  -DZEROBUS_FFI_LIBRARY="$PWD/lib/libzerobus_ffi.a" \
  -DZEROBUS_FFI_HEADER_DIR="$PWD/lib"
cmake --build build -j

Windows (PowerShell) では、代わりに .lib アーカイブを指定してください：

PowerShell

cmake -S cpp -B build `
  -DZEROBUS_FFI_LIBRARY="$PWD/lib/zerobus_ffi.lib" `
  -DZEROBUS_FFI_HEADER_DIR="$PWD/lib"
cmake --build build -j

独自のCMakeプロジェクトでソースチェックアウトからSDKをビルドするには、それをサブディレクトリとして追加し、ターゲットをLinkします。FetchContent を使用して構成時に取得することもできます。これはRustソースからFFIを構築するため、Rustツールチェーンが必要です：

Cmake

add_subdirectory(path/to/zerobus-sdk/cpp)
target_link_libraries(your_app PRIVATE zerobus::zerobus)

代わりに add_subdirectory を通じてビルド済みのバンドルを使用するには、まず FFI パスを設定して、存在しない Rust ソースからビルドしようとするのではなく、CMake がバンドルされたアーカイブにリンクするようにします。Windows で zerobus_ffi.lib を使用します：

Cmake

set(ZEROBUS_FFI_LIBRARY "/path/to/bundle/lib/libzerobus_ffi.a")
set(ZEROBUS_FFI_HEADER_DIR "/path/to/bundle/lib")
add_subdirectory(path/to/zerobus-sdk/cpp zerobus-cpp)
target_link_libraries(your_app PRIVATE zerobus::zerobus)

JSONの例:

取り込みは非同期かつパイプライン化されています。ingest_* メソッドはレコードをキューに入れ、すぐに戻ります。各レコードの後に待機するのではなく、バッチをキューに入れ、flush() を一度呼び出します。

Cpp

#include "zerobus/zerobus.hpp"
#include <string>
#include <vector>

int main() {
  // See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
  const std::string SERVER_ENDPOINT = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com";
  const std::string DATABRICKS_WORKSPACE_URL = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com";
  const std::string TABLE_NAME = "main.default.air_quality";
  const std::string CLIENT_ID = "your-client-id";
  const std::string CLIENT_SECRET = "your-client-secret";

  zerobus::Sdk sdk = zerobus::Sdk::builder()
                         .endpoint(SERVER_ENDPOINT)
                         .unity_catalog_url(DATABRICKS_WORKSPACE_URL)
                         .application_name("my-app")
                         .build();

  zerobus::TableProperties table;
  table.table_name = TABLE_NAME;   // empty descriptor => JSON stream

  zerobus::StreamOptions options;
  options.record_type = zerobus::RecordType::Json;

  zerobus::Stream stream =
      sdk.create_stream(table, CLIENT_ID, CLIENT_SECRET, options);

  std::vector<std::string> batch = {
      R"({"device_name": "sensor-001", "temp": 20, "humidity": 60})",
      R"({"device_name": "sensor-002", "temp": 22, "humidity": 55})",
  };
  stream.ingest_json_records(batch);   // queue the batch — no per-record wait
  stream.flush();                      // wait once for all acks
  stream.close();

  return 0;
}

失敗するたびに zerobus::ZerobusException がスローされ、メッセージと is_retryable() フラグが保持されます。ブロッキングなしで継続的なストリームの耐久性を追跡するには、StreamOptions::ack_callback を介して AckCallback を登録します。コールバックはバックグラウンドスレッドでシリアル化されて実行されるため、noexcept である必要があります。スレッド処理、ドレインポリシー、およびライフタイムコントラクトの詳細については、C++ SDK ドキュメントを参照してください。

型安全な取り込みのために、Protocol Buffersを次の2つの方法のいずれかで利用できます：

• ProtoSchema::from_uc_json() を使用してUnity Catalogからスキーマを生成します。これはテーブルのメタデータから直接記述子とJSON-to-protoエンコーダーを構築するため、.proto ファイルや protoc は不要です：

  1. Get a table API (GET /api/2.1/unity-catalog/tables/{full_name}) からテーブルのメタデータJSONを取得します。Service Principalには、テーブルに対する SELECT が必要です。
  2. メタデータを ProtoSchema::from_uc_json() に渡して、記述子とエンコーダーをビルドします。
  3. TableProperties::descriptor_proto を設定してから、ingest_proto_records() で取り込みます。

• コンパイル時の型指定のために、チェックインされた .proto を protoc でコンパイルします。

実行可能なウォークスルーについては、Protocol Buffersの例を参照してください。

同一のgRPC接続を介したApache Arrowレコードバッチの列指向またはバッチ指向の取り込みについては、Use Arrow Flight with Zerobus Ingestを参照してください。

完全なドキュメント、構成オプション、バッチ取り込み、および Protocol Buffer の例については、C++ SDK リポジトリを参照してください。

C# SDK

備考

ベータ版

C# / .NET SDK はベータ版です。Databricks.Zerobus パッケージはプレリリース版です。

.NET 8.0 以上が必要です。この SDK は、ネイティブな gRPC ストリーミング、OAuth、および自動復旧を提供します。シンプルなセットアップには JSON を、本番運用ワークロードには Protocol Buffers をサポートしています。Arrow Flight は C# SDK では使用できません。

Databricks.Zerobus パッケージをプロジェクトに追加します。

Bash

dotnet add package Databricks.Zerobus

JSONの例:

Csharp

using Databricks.Zerobus;

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
const string SERVER_ENDPOINT = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com";
const string DATABRICKS_WORKSPACE_URL = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com";
const string TABLE_NAME = "main.default.air_quality";
const string CLIENT_ID = "your-client-id";
const string CLIENT_SECRET = "your-client-secret";

using var sdk = ZerobusSdk.CreateBuilder()
    .Endpoint(SERVER_ENDPOINT)
    .UnityCatalogUrl(DATABRICKS_WORKSPACE_URL)
    .Build();

using var stream = sdk.CreateJsonStream(TABLE_NAME, CLIENT_ID, CLIENT_SECRET);

long offset = stream.IngestRecord(
    """{"device_name": "sensor-1", "temp": 22, "humidity": 55}""");
stream.WaitForOffset(offset);
stream.Close();

IngestRecord レコードのoffsetを返し、そのレコードが永続化されるまでWaitForOffsetがブロックします。バッチを取り込むにはIngestRecordsを使用します。これはレコードの配列を受け取り、最後のoffsetを返します。offsetでのブロックはオプションです。「Message blocking and acknowledgment」を参照してください。

Protocol Buffers: 型安全な取り込みを行うには、sdk.CreateProtoStream(TABLE_NAME, descriptorProto, CLIENT_ID, CLIENT_SECRET) でストリームを作成します。ここで、descriptorProto はコンパイル済みメッセージのシリアル化された DescriptorProto バイトであり、その後 stream.IngestRecord(protoBytes) で取り込みを行います。

完全なドキュメント、構成オプション、およびProtocol Bufferの例については、C# SDKリポジトリを参照してください。

TypeScript SDK

Node.js 16 以上が必要です。SDK は、JavaScript Promises を介した非同期サポートにより、高いパフォーマンスを提供します。JSON (最もシンプル) と Protocol Buffers (本番運用に推奨) をサポートしています。

Bash

npm install @databricks/zerobus-ingest-sdk

JSONの例:

TypeScript

import { ZerobusSdk, RecordType } from '@databricks/zerobus-ingest-sdk';

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
const SERVER_ENDPOINT = 'https://1234567890123456.zerobus.us-west-2.cloud.databricks.com';
const DATABRICKS_WORKSPACE_URL = 'https://dbc-a1b2c3d4-e5f6.cloud.databricks.com';
const TABLE_NAME = 'main.default.air_quality';
const CLIENT_ID = 'your-client-id';
const CLIENT_SECRET = 'your-client-secret';

const sdk = new ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL);

const stream = await sdk.createStream({ tableName: TABLE_NAME }, CLIENT_ID, CLIENT_SECRET, {
  recordType: RecordType.Json,
});

try {
  for (let i = 0; i < 100; i++) {
    const record = { device_name: `sensor-${i}`, temp: 22, humidity: 55 };
    await stream.ingestRecordOffset(record);
  }
} finally {
  await stream.close();
}

プロトコル バッファー: 型セーフな取り込みを行うには、 RecordType.Proto (デフォルト) でプロトコル バッファーを使用し、テーブル プロパティにdescriptorProtoを指定します。

Arrow Flight (ベータ版): 同じ gRPC 接続を介して Apache Arrow RecordBatchデータを列指向またはバッチ指向で取り込むには、 「Arrow Flight を Zerobus Ingest で使用する」を参照してください。

完全なドキュメント、構成オプション、バッチ取り込み、およびプロトコル バッファーの例については、 TypeScript SDK リポジトリを参照してください。

REST API

REST API を使用すると、 /zerobus/v1/tables/<table-name>/insertエンドポイントに HTTP POST リクエストを送信して、単一のレコードを取り込むことができます。レコード自体はリクエスト本文に含まれており、JSON 形式である必要があります。

この例では、CURL を使用して REST API で Zerobus Ingest にデータをプッシュする方法について説明します。

ヘッダー

リクエストを正しく認証してフォーマットするには、2 つの特定の HTTP ヘッダーが必要です。

• Content-Type: application/json

  • コンテンツ タイプを指定するための必須フィールド。現在、サポートされているメッセージ形式は JSON のみです。

• Authorization: Bearer <token>

  • <token>後ほど提供される curl コマンドを使用して取得した OAuth トークンに置き換えます。

OAuth取得: これらは 1 時間ごとに期限切れになるため、更新する必要があります。 OAuth トークンを再度取得することで更新できます。

次の事項を記入してください:

• $CATALOG、$SCHEMA、$TABLE、$WORKSPACE_ID、 $WORKSPACE_URL
• $DATABRICKS_CLIENT_ID そして $DATABRICKS_CLIENT_SECRET
  • これら 2 つのパラメーターは、作成したサービス プリンシパルに対応します。

Bash

authorization_details=$(cat <<EOF
[{
  "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": "$CATALOG.$SCHEMA"
},
{
  "type": "unity_catalog_privileges",
  "privileges": ["SELECT", "MODIFY"],
  "object_type": "TABLE",
  "object_full_path": "$CATALOG.$SCHEMA.$TABLE"
}]
EOF
)

export OAUTH_TOKEN=$(curl -X POST \
  -u "$DATABRICKS_CLIENT_ID:$DATABRICKS_CLIENT_SECRET" \
  -d "grant_type=client_credentials" \
  -d "scope=all-apis" \
  -d "resource=api://databricks/workspaces/$WORKSPACE_ID/zerobusDirectWriteApi" \
  --data-urlencode "authorization_details=$authorization_details" \
  "$WORKSPACE_URL/oidc/v1/token" | jq -r '.access_token')

レコードの取り込み:

次の事項を記入してください:

• $ZEROBUS_ENDPOINT

  • 「ワークスペース URL と Zerobus 取り込みエンドポイントを取得する」セクションで定義されているとおりです。

• $CATALOG、$SCHEMA、$TABLE、$WORKSPACE_ID、 $WORKSPACE_URL

• $OAUTH_TOKEN

  • これは前のステップで作成されました。

リクエスト本体は JSON オブジェクトのリストである必要があります。

Bash

curl -X POST \
  "$ZEROBUS_ENDPOINT/zerobus/v1/tables/$CATALOG.$SCHEMA.$TABLE/insert" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OAUTH_TOKEN" \
  -d '[{ "device_name": "device_num_1", "temp": 28, "humidity": 60 },
       { "device_name": "device_num_1", "temp": 28, "humidity": 60 }]'

すべての情報が正しく入力されている場合は、HTTP ステータス コード 200 の空の JSON 応答が返されます。

エラーに対応する

上記の例は、正常系（ハッピーパス）を示しています。本番運用では、取り込み処理をエラーハンドリングでラップしてください。SDK は、組み込みのリカバリ機能を通じて、ネットワークの問題などの一時的なエラーを自動的に再試行します。無効な認証情報やテーブルの欠落など、回復不可能な障害は ZerobusException として表面化します：

Python

from zerobus.sdk.shared import ZerobusException

try:
    stream.ingest_record_offset(record)
except ZerobusException as e:
    # Handle the failure: log it, fix the cause, recover on a new stream, or stop.
    ...

SDK は一時的な障害からも自動的に回復し、ストリームが完全に失敗した場合には未確認のレコードを救出できるようにします。レジリエントクライアントパターンと完全なエラーリファレンスについては、回復および再試行パターンとZerobus Ingest のエラー処理を参照してください。

次のステップ

• Zerobus Ingest のクォータを確認する
• データファイルのレイアウトを最適化する
• Zerobus レスキュー列