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

Arrow FlightをZerobus Ingestで使用する

Arrow Flight 取り込みを使用すると、すべての行を最初に JSON または Protocol Buffers (protobuf) に変換することなく、Apache Arrow RecordBatch データを直接 Zerobus Ingest に送信できます。これは、Arrow Flight をサポートする Zerobus SDK において、JSON および protobuf に続く3番目のレコード形式オプションであり、同じ gRPC 接続上で実行されます。これは、同じ Zerobus Endpoint、同じ OAuth フロー、および同じ x-databricks-zerobus-table-name ヘッダー規則を使用します。ワイヤプロトコルは Arrow Flight DoPut であり、gRPC を介して Arrow IPC メッセージを伝送します。

Arrow Flightはいつ使うべきか​

Arrow Flightは、以下のシナリオに最適です。

  • ご使用のアプリケーションは、pyarrow.Table や pyarrow.RecordBatch（Python）、arrow-rs クレート（Rust）からの arrow_array::RecordBatch、または VectorSchemaRoot（Java）など、すでに Arrow データを生成しています。Polars や DataFusion など、Arrow 上に構築された DataFrame ライブラリは、このパスに自然に適合します。
  • レコードを1件ずつ送信するのではなく、行をバッチ処理で取り込みます。
  • スキーマが幅広く、数値が多く、またはアナリティクス指向であり、行ごとのシリアル化により CPU オーバーヘッドが顕著に増加します。
  • あなたは、短時間のデータを集約し、それを1列形式のバッチとして送信するコレクターまたはゲートウェイを構築しています。

Arrow Flight は通常、スパースで1行ずつ処理するトラフィックには最適ではありません。そのような場合、通常は SDK gRPC パス経由の JSON または protobuf の方がシンプルです。「インターフェースの選択」を参照してください。

取り込みモデルの仕組み​

Arrow Flightのデータ取り込みでは、1つのストリームが1つのターゲットテーブルに書き込みます。データを取り込むには、以下の手順に従ってください。

  1. 宛先Deltaテーブルのスキーマと一致するArrowスキーマを定義します。
  2. そのテーブルの Zerobus Arrow ストリームを開きます。
  3. RecordBatch （またはTable ）のペイロードを送信します。
  4. 最後のオフセットを待つか、 flush()を呼び出して耐久性を確認します。
  5. ストリームを閉じます。

Zerobus SDK を使用する場合、SDK が低レベルの Arrow Flight ワイヤの詳細を処理します。これは、ArrowデータをIPC形式にシリアル化し、大きなバッチをより小さく順序付けられたFlightメッセージに自動的に分割します。

サーバーは、レコードが永続化されるにつれて累積的な進捗状況を報告します。したがって、送信中に障害が発生した場合、大きな論理バッチが部分的に永続化される可能性があります。ingest_batch() は送信されたバッチに対して引き続き 1 つの論理オフセットを返し、そのオフセットを待機することで、すべてのレコードが永続化されたことが確認されます。Arrow Flight バッチは例外ですを参照してください。

Zerobusは、ArrowフィールドとDelta列を名前で照合します。Arrowスキーマには、必須（NULL非許容）のDelta列がすべて含まれている必要があります。NULL許容列は省略できます。その場合、ZerobusはNULLとして書き込みます。ターゲットテーブルに存在しないフィールドは含めないでください。含まれるフィールドは、Deltaスキーマの相対的な順序に従い、そのNULL値の許容と一致し、ターゲットカラムの型を使用する必要があります。詳細については、 スキーマ照合ルール を参照してください。

Arrow バッチは、10 MB の gRPC メッセージサイズ制限の対象外です。ただし、RecordBatch 内の各行は 10 MB の制限内に収まる必要があります。SDK は大きなバッチを自動的に複数のワイヤーメッセージに分割しますが、サイズ超過した 1 行を分割することはできません。Zerobus Ingest のクォータを参照してください。

クライアントを記述する​

以下の例では、Python SDK および Rust SDK を使用しています。Arrow Flight をサポートするその他の言語については、Zerobus SDK リポジトリを参照してください。

Python SDK は、ストリーム作成時にpyarrow.Schemaを、各取り込み呼び出し時にpyarrow.RecordBatchまたはpyarrow.Tableを受け入れます。

Bash
pip install "databricks-zerobus-ingest-sdk[arrow]"
Python
import pyarrow as pa

from zerobus.sdk.sync import ZerobusSdk

# See "Get your workspace URL and Zerobus Ingest endpoint" in zerobus-ingest.md.
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"

schema = pa.schema(
    [
        ("device_name", pa.large_utf8()),
        ("temp", pa.int32()),
        ("humidity", pa.int64()),
    ]
)

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

stream = sdk.create_arrow_stream(TABLE_NAME, schema, CLIENT_ID, CLIENT_SECRET)

try:
    for start in range(0, 10_000, 1_000):
        end = start + 1_000
        batch = pa.record_batch(
            {
                "device_name": [f"sensor-{i}" for i in range(start, end)],
                "temp": [20 + (i % 5) for i in range(start, end)],
                "humidity": [55 + (i % 10) for i in range(start, end)],
            },
            schema=schema,
        )
        stream.ingest_batch(batch)
    stream.flush()
finally:
    stream.close()

stream.ingest_batch() もpyarrow.Tableを受け入れます。SDKは、送信前に内部で単一のRecordBatchに変換します。各呼び出しは、論理オフセットを返します。この例では、いくつかのバッチを作成し、flush()を1回呼び出して、保留中のすべてのバッチが永続化されていることを確認します。続行する前に特定のバッチを確認する必要がある場合は、wait_for_offset()を使用します。最後のオフセットを待機すると、それより前のすべてのオフセットも確認されます。待機するタイミングと確認応答の仕組みについては、メッセージのブロックと確認応答を参照してください。

VARIANT列の取り込み​

Apache Arrow にはネイティブの VARIANT 型はありません。Arrow Flight を介して VARIANT 列に取り込むには、その列の基盤となる metadata フィールドと value フィールドを 2 つの LargeBinary 列の構造体として構築し、その構造体を RecordBatch に含めます。gRPC SDK および REST を介する場合は、代わりに Variant 値を JSON エンコードされた文字列として渡します。「サポートされているデータ型」を参照してください。

次の Rust の例では、JSON 行から VARIANT 構造体列を構築し、それを取り込みます。

Rust
fn variant_struct(json_rows: &[&str]) -> ArrayRef {
    let mut metas: Vec<Vec<u8>> = Vec::new();
    let mut vals: Vec<Vec<u8>> = Vec::new();
    for json in json_rows {
        let mut vb = VariantBuilder::new();
        vb.append_json(json).expect("invalid JSON for variant");
        let (metadata, value) = vb.finish();
        metas.push(metadata);
        vals.push(value);
    }
    let fields = Fields::from(vec![
        Field::new("metadata", DataType::LargeBinary, false),
        Field::new("value", DataType::LargeBinary, false),
    ]);
    let meta_arr = Arc::new(LargeBinaryArray::from_iter_values(metas)) as ArrayRef;
    let val_arr = Arc::new(LargeBinaryArray::from_iter_values(vals)) as ArrayRef;
    Arc::new(StructArray::try_new(fields, vec![meta_arr, val_arr], None).expect("variant struct"))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client_id = std::env::var("DATABRICKS_CLIENT_ID")?;
    let client_secret = std::env::var("DATABRICKS_CLIENT_SECRET")?;

    let variant_type = DataType::Struct(Fields::from(vec![
        Field::new("metadata", DataType::LargeBinary, false),
        Field::new("value", DataType::LargeBinary, false),
    ]));
    let schema = Arc::new(ArrowSchema::new(vec![
        Field::new("id", DataType::Int32, true),
        Field::new("payload", variant_type, true),
    ]));

    let sdk = ZerobusSdk::builder()
        .endpoint(ENDPOINT)
        .unity_catalog_url(UC_URL)
        .build()?;

    let mut stream = sdk
        .stream_builder()
        .table(TABLE)
        .oauth(&client_id, &client_secret)
        .arrow(schema.clone())
        .ipc_compression(None)
        .build_arrow()
        .await?;

    let ids = Int32Array::from(vec![1, 2, 3]);
    let payload = variant_struct(&[
        r#"{"user":"alice","tags":[1,2,3]}"#,
        r#""just a string""#,
        r#"{"nested":{"a":true,"b":null,"c":3.14}}"#,
    ]);
    let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(ids) as ArrayRef, payload])?;

    let offset = stream.ingest_batch(batch).await?;
    stream.flush().await?;
    stream.close().await?;
    Ok(())
}

この例はRustで記述されています。他の言語での同等の使用方法については、Zerobus SDKリポジトリを参照してください。

IPC圧縮​

デフォルトでは、Arrow IPCペイロードは圧縮されずに送信されます。2つのコーデックのいずれかを使用して、転送中にオプションで圧縮することも可能です。

  • LZ4_FRAME：高速、低CPUオーバーヘッド、控えめな圧縮率です。クライアントのCPUに制約があるものの、通信量を削減したい際はこちらをお勧めします。
  • ZSTD：より高い圧縮率、バッチあたりのCPUが増加しました。クライアントが追加のCPUコストを負担できる場合に、有効にしてください。

圧縮はワイヤー上のバイト数を削減しますが、クライアント側でCPUコストが増加します。ペイロードが小さいほど、ネットワークのボトルネックを回避し、ネットワークコストを削減できます。

ArrowStreamConfigurationOptions の ipc_compression フィールドを設定します：

Python
from zerobus.sdk.shared.arrow import IPCCompression, ArrowStreamConfigurationOptions

options = ArrowStreamConfigurationOptions(ipc_compression=IPCCompression.ZSTD)
stream = sdk.create_arrow_stream(
    TABLE_NAME, schema, CLIENT_ID, CLIENT_SECRET, options=options
)

ベストプラクティス​

Arrow Flightの吸入機能から最高のパフォーマンスと信頼性を得るには、以下のガイドラインに従ってください。

  • バッチごとに新しいストリームを開くのではなく、多くのバッチにストリームを再利用します。 ストリームの作成にはかなりのオーバーヘッドがかかりますが、多くのバッチにわたってストリームを再利用することで軽減できます。
  • 1回のバッチ処理で複数の行を送信します。1回の呼び出しで1行ずつ処理するのではなく、アプリケーションのサイズに合わせた自然なバッチ処理から始めましょう。一度に1行ずつ送信する方法は機能しますが、Arrowを使用する際のパフォーマンス上の利点のほとんどが失われます。
  • 管理されたチェックポイントでflush()を呼び出してください。これにより、個々のバッチをブロックすることなく、バッチグループ全体の明確な耐久性境界を設定できます。
  • throughputを向上させるには、IPC圧縮を有効にしてください。クライアントにCPUの余裕がある場合、ほとんどのワークロードにはZSTDが推奨されます。クライアントのCPUリソースが制限されている場合は、LZ4_FRAMEを使用するか、圧縮なしを選択してください。
  • プロデューサーがすでに列指向である場合は、Arrow Flight を使用してください。ソースデータが本来行指向で小規模な場合は、JSON または protobuf で Zerobus Ingest を使用する方が単純な場合が多いです。Zerobus Ingest の使用を参照してください。

エラー処理と回復​

Arrow Flight ストリームは、Zerobus Ingest の他の部分と同じ gRPC エラーカテゴリを使用します。エラーコード、再試行に関するガイダンス、およびクライアントとサーバーの完全な分類については、 「Zerobus Ingest エラー処理」を参照してください。

自動リカバリ（default）で SDK を設定すると、一時的な障害が発生した際に SDK は透過的に再接続し、未承認のバッチをリプレイします。ストリームが未承認の作業を残したまま終了した後、SDK はクライアントが受け入れたもののサーバーが承認しなかったバッチ（まだ送信されていない可能性のあるバッチを含む）を保持します。

自動リカバリが終了したら、失敗の原因を修正し、close() を呼び出してストリームの未確認バッチをファイナライズします。ストリームはすでに失敗しているため、ファイナライズが成功しても close() が同じ終了エラーを返す可能性があります。

get_unacked_batches() は、ストリームが閉じられた後にのみ呼び出すことができます。永続化またはアプリケーション管理によるリプレイのために、保持されたバッチを返します。代替ストリームの作成方法、バッチの永続化方法、およびそれらの再試行方法は、アプリケーションのリカバリポリシーによって異なります。

Python
from zerobus.sdk.shared import ZerobusException

try:
    stream.close()
except ZerobusException:
    # The terminal error can be returned after closure is finalized.
    pass

unacked_batches = stream.get_unacked_batches()

その他のリソース​

  • Zerobus Ingest の使用: Zerobus Ingest をまだ設定していない場合は、ここから起動して、ワークスペース URL の検索、ターゲット Delta テーブルの作成、および Service Principal の構成に関する手順を確認してください。これらのステップは、すべてのレコードフォーマットで共通です。
  • Zerobus Ingestのクォータ：本番運用にデプロイする前に、Zerobusのクォータを確認してください。throughput、レイテンシ、およびパーティション分割テーブルの制限はすべて、Arrow Flightに適用されます。
  • Zerobus Ingest のエラー処理：gRPC エラーコードの完全なリスト、およびクライアントに推奨される再試行と回復の動作については、このページを参照してください。
このページの見出し