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

Arrow FlightをZerobus Ingestで使用する

備考

ベータ版

Arrow Flight インジェクションはベータ版です。

Arrow Flight 取り込みを使用すると、すべての行を最初に JSON または Protocol Buffers (protobuf) に変換することなく、Apache Arrow RecordBatch データを直接 Zerobus Ingest に送信できます。これは、JSON および protobuf に続く Zerobus SDK の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 フォーマットにシリアル化し、大きなバッチを小さな転送メッセージに分割します。サーバーはこれらのメッセージを個別に確認応答します。

Arrow Flight は、論理バッチ全体に対してオール・オア・ナッシングの耐久性を提供するものではありません。Arrow バッチは非常に大きくなる可能性があり、SDK はそれを個別の転送メッセージに分割して受信確認を行うため、途中で障害が発生した場合、大きなバッチは部分的に永続化される可能性があります。これは、アトミックに commit され、10 MB のメッセージサイズに制限される JSON および protobuf バッチとは異なります。Arrow Flight バッチは例外ですを参照してください。

論理オフセットの抽象化は、このチャンク化の上でも維持されます。ingest_batch() は送信したバッチに対して単一の論理オフセットを返し、そのオフセットに対する wait_for_offset() は、バッチを構成するすべての転送メッセージが確認された後にのみ完了します。（メソッド名は Python SDK のものです。他の SDK では、Java の ingestBatch や waitForOffset など、同等のメソッドが公開されています。）

Protobuf スキーマ ルールと同様に、ストリームに渡すスキーマはターゲットのDeltaテーブルに適合している必要があります。つまり、少なくともすべてのNULL非許容列が含まれている必要があります。スキーマでは、Deltaテーブルに存在するNULL許容列を省略できます（これは破壊的でないスキーマ変更として扱われます）が、その他の不一致はすべて拒否されます。各 Arrow フィールドの型は、その Delta 列と互換性がある必要があります。サポートされているDelta型については、サポートされているデータ型を参照してください。

SDK は大きなバッチをトランスポートメッセージに分割するため、Arrow バッチは JSON や Protocol Buffers バッチのように 10 MB のメッセージサイズ制限の対象にはなりません。Arrow Flight は同じ gRPC トランスポート上で実行されるため、同じ throughput、レイテンシー、およびクォータの特性が適用され、これらはすべてより高いワークロードに対応するようにスケーリングされます。Zerobus Ingest のクォータを参照してください。

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

以下の例では、Zerobus Ingest の使用例で使用されているのと同じ air_quality テーブルに対して Arrow Flight ストリームを開きます。簡潔にするため Python と Rust で示していますが、同じビルダー、構成オプション、呼び出しシーケンスがすべての Zerobus SDK で利用可能です。使用する言語に合わせて構文を調整し、言語固有の Arrow 型については SDK リポジトリを参照してください。

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

Bash
pip install "databricks-zerobus-ingest-sdk[arrow]" pyarrow
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)

row_count = 1_000
batch = pa.record_batch(
    {
        "device_name": [f"sensor-{i}" for i in range(row_count)],
        "temp": [20 + (i % 5) for i in range(row_count)],
        "humidity": [55 + (i % 10) for i in range(row_count)],
    },
    schema=schema,
)

try:
    offset = stream.ingest_batch(batch)

    # Optional: block until the batch is durably written
    stream.wait_for_offset(offset)
finally:
    stream.close()

stream.ingest_batch() もpyarrow.Tableを受け入れます。SDKは、送信前に内部で単一のRecordBatchに変換します。各呼び出しは、論理オフセットを返します。オフセットでのブロックはオプションです。待機するタイミングと確認応答の仕組みについては、メッセージのブロックと確認応答を参照してください。

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コストが増加します。ペイロードが小さいほど、ネットワークのボトルネックを回避し、ネットワークコストを削減できます。

Python SDKで、 ArrowStreamConfigurationOptionsのipc_compressionフィールドを設定します。

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

options = ArrowStreamConfigurationOptions(ipc_compression=IPCCompression.ZSTD)

Rust SDKでは、ビルダーで設定します。CompressionType列挙型はarrow-ipcクレートに含まれているので、依存関係として追加してください。

Bash
cargo add arrow-ipc
Rust
use arrow_ipc::CompressionType;

let stream = sdk
    .stream_builder()
    .table(TABLE_NAME)
    .oauth(CLIENT_ID, CLIENT_SECRET)
    .arrow(schema)
    .ipc_compression(Some(CompressionType::ZSTD))
    .build_arrow()
    .await?;

ベストプラクティス​

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 エラー処理」を参照してください。

SDKを自動復旧（デフォルト設定）で構成すると、一時的な障害発生時に透過的に再接続し、未確認のバッチを再再生します。ストリームが終了した後でも、サーバーが受信したがまだ確認応答していないバッチを取得できます。これは、ストリームが正常に終了した場合でも、回復不能な障害によって終了した場合でも同様に適用されます。Python SDKでは：

Python
# Retry unacked_batches against a freshly created stream
if stream.is_closed:
    unacked_batches = stream.get_unacked_batches()

Rust SDKでは、 stream.get_unacked_batches().await?を呼び出して、再試行のために未確認のバッチを取得します。

その他のリソース​

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