Typy komunikatów

Gdy pobierasz pliki z SDK Zerobus Ingest przez gRPC, wybierasz, jak rekordy są kodowane w danym strumieniu. Zerobus Ingest obsługuje cztery formaty wiadomości, dzięki czemu możesz wybierać kompromis między prostotą, bezpieczeństwem typów i przepustowością dla swojego obciążenia: JSON, Bufory Protokołu (protobuf), Apache Arrow i Avro (Beta). Każdy format jest weryfikowany względem schematu tabeli Delta, zanim dane staną się trwałe. Zobacz zarządzanie schematami.

Diagram pokazuje JSON, protobuf i Apache Arrow. Nie obejmuje Avro, które używa zwartego, wierszowego kodowania podobnego do protobuf.

Jakiego formatu powinieneś użyć?

Poniższe fragmenty pokazują kształt każdego formatu w Python SDK. Zakładają, że już stworzyłeś klienta SDK i znasz swoją docelową tabelę. Pełną konfigurację (endpoint, tabela, nazwa główna usługi) i przykłady we wszystkich językach znajdziesz w Use Zerobus Ingest.

JSON

JSON to najprostszy sposób na początek: wysyłasz rekordy jako obiekty JSON bez żadnej definicji schematu do kompilacji. Jest idealny do uruchamiania, prototypowania i produkcji tam, gdzie wygoda jest ważniejsza niż surowa przepustowość. Dla dużych nakładów produkcyjnych bardziej efektywny jest format binarny (protobuf lub Arrow).

Utwórz strumień dla rekordów JSON, przekazując nazwę tabeli do TableProperties bez deskryptoru:

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

stream.ingest_record_offset({"device_name": "sensor-1", "temp": 22, "humidity": 55})

Pełny przewodnik JSON znajdziesz w artykule Write a client.

Bufory protokołu

Protobuf zapewnia bezpieczne dla typów, kompaktowe kodowanie binarne i jest zalecanym formatem dla większości produkcyjnych, wierszowych obciążeń. Definiujesz schemat protobuf dopasowany do docelowej tabeli Delta (zobacz Protobuf schema), kompilujesz go, a pakiet SDK przyjmuje rekordy rekord po rekordzie przez gRPC.

Korzystanie z protobuf wymaga trzech kroków: wygenerowania schematu .proto odpowiadającego Twojej tabeli, skompilowania go do modułu językowego, a następnie pobierania rekordów, przekazując deskryptor do TableProperties. Poniższy przykład wykorzystuje Python SDK.

1. Wygeneruj .proto schemat ze swojej tabeli. Python SDK zawiera narzędziegenerate_proto, które odczytuje Twoją tabelę Delta i tworzy odpowiadający schemat:

python -m zerobus.tools.generate_proto \
    --uc-endpoint "https://<workspace-id>.cloud.databricks.com" \
    --client-id "<client-id>" \
    --client-secret "<client-secret>" \
    --table "main.default.air_quality" \
    --output "record.proto" \
    --proto-msg "AirQuality"

Wygenerowany schemat wykorzystuje proto2 składnię, z opcjonalnym polem dla każdej kolumny Delta:

syntax = "proto2";
message AirQuality {
    optional string device_name = 1;
    optional int32 temp = 2;
    optional int64 humidity = 3;
}

2. Skompiluj schemat do modułu Python za pomocą kompilatora protobuf:

pip install "grpcio-tools>=1.60.0,<2.0"
python -m grpc_tools.protoc --python_out=. --proto_path=. record.proto

To generuje record_pb2.py.

3. Importowanie rekordów przez przekazanie skompilowanego deskryptora do TableProperties (co jest ustawieniem domyślnym dla protobuf). SDK używa deskryptora do serializacji każdego rekordu:

import record_pb2

descriptor_bytes = record_pb2.AirQuality.DESCRIPTOR.file.serialized_pb
table_properties = TableProperties(TABLE_NAME, descriptor_bytes)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)

record = record_pb2.AirQuality(device_name="sensor-1", temp=22, humidity=55)
stream.ingest_record_offset(record)

Powyższy przykład wykorzystuje Python SDK. Narzędzia różnią się w zależności od języka: niektóre SDK zawierają narzędzie generate_proto, które generuje .proto na podstawie tabeli, podczas gdy inne (takie jak Go i TypeScript) kompilują istniejący .proto. Kroki dotyczące poszczególnych języków można znaleźć w notatkach protobufowych w każdej zakładce SDK w Write a client. Źródła narzędzi i pełne przykłady można znaleźć w repozytorium SDK Zerobus.

Strzała Apache

Pozyskiwanie danych Apache Arrow wysyła dane ArrowRecordBatch bezpośrednio przez to samo połączenie gRPC, zamiast najpierw konwertować każdy wiersz do formatu JSON lub protobuf. To najlepszy wybór, gdy aplikacja już generuje dane Arrow lub gdy przetwarzasz wiersze partiami, zwłaszcza w przypadku szerokich schematów, schematów zdominowanych przez dane liczbowe lub schematów ukierunkowanych na analitykę, gdzie serializacja wiersz po wierszu wprowadza dodatkowy narzut.

Arrow też dobrze sprawdza się w bardzo dużych partiach. W przeciwieństwie do metod wsadowych JSON i protobuf, które działają na zasadzie „wszystko albo nic” i są ograniczone limitem rozmiaru pojedynczej wiadomości, mechanizm Arrow Flight dzieli duży wsad na mniejsze komunikaty transportowe, które są wysyłane i potwierdzane indywidualnie. Zobacz Partie Arrow Flight stanowią wyjątek i Używanie Arrow Flight z Zerobus Ingest.

Otwórz strumień Arrow za pomocą pyarrow.Schema i pozyskaj dane RecordBatch:

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

stream.ingest_batch(batch)

Pełny przewodnik Arrow Flight, w tym definicję schematu, grupowanie i kompresję, znajdziesz w artykule Use Arrow Flight with Zerobus Ingest.

Avro

Apache Avro to zwarty, binarny format zorientowany na wiersze, sterowany schematem. Podobnie jak protobuf, każdy rekord jest kodowany względem schematu, zamiast zawierać nazwy pól bezpośrednio w rekordzie, co sprawia, że rekordy są małe. Różnica polega na tym, że protobuf używa skompilowanego .proto deskryptora, podczas gdy Avro opisuje swoje rekordy za pomocą schematu napisanego w JSON. Pozyskiwanie danych Avro przebiega przez gRPC przez SDK Zerobus. Informacje o tym, które zestawy SDK obsługują Avro, znajdziesz w repozytorium SDK Zerobus.

Avro dobrze pasuje, gdy producent już pracuje z Avro, na przykład gdy czyta dane Avro z systemu nadrzędnego lub posiada schemat zapisu Avro.

Obsługa schematów

Zerobus Ingest nie korzysta z rejestru schematów, a rekordy w transmisji nie zawierają identyfikatora schematu ani odcisku palca. Zamiast tego podajesz schemat zapisujący Avro jako JSON podczas tworzenia strumienia, a każdy rekord w tym strumieniu jest dekodowany względem tego schematu. Schemat jest weryfikowany względem docelowej tabeli Delta na początku, więc niekompatybilny schemat jest odrzucany przed jakimkolwiek przepływem danych.

Schemat modułu zapisującego jest ustalony na cały okres życia strumienia. Aby pozyskiwać dane z innym schematem, stwórz nowy strumień.

Pola Avro są łączone z kolumnami Delta według nazwy na każdym poziomie, więc kolejność pól w schemacie writera nie ma znaczenia. Aby dowiedzieć się, jak typy Avro mapują się na kolumny Delta i które kolumny nie mogą być targetowane, zobacz Avro types. Aby dowiedzieć się, jak rekordy są weryfikowane względem tabeli, zobacz zarządzanie schematami.

Schemat zapisu Avro musi deklarować każdy typ inline. Odwołania do typów nazwanych, które ponownie wykorzystują typ record, enum lub fixed nazwany w innym miejscu schematu, odwołując się do jego nazwy, nie są obsługiwane.

Jak dostarczać dane

Niezależnie od tego, czy używasz SDK, możesz pozyskiwać jedną wartość na raz lub w partiach. Różnica polega na tym, kto koduje dane do Avro:

• Dzięki SDK możesz:
  • Pozwól, aby SDK zakodował Twoje dane: podaj SDK swoje rekordy Avro, a on zakoduje je zgodnie ze schematem zapisującego.
  • Wysyłaj wcześniej zakodowane bajty Avro: jeśli już przechowujesz dane zakodowane w Avro, na przykład z systemu upstream, SDK przesyła je w niezmienionej postaci.
• Bez SDK, wywołując bezpośrednio API gRPC, wysyłasz tylko prekodowane bajty Avro. Stwórz strumień za pomocą schematu writera, a następnie wyślij surowe kodowanie binarne Avro swoich danych, bez nagłówka pliku container, prefiksu schema-registry ID czy odcisku palca.

Napisz do klienta

Poniższe przykłady wykorzystują SDK Python i Rust. Tworzą strumień, wysyłają rekordy kodowane przez SDK względem schematu zapisującego i opróżniają bufor. Zakłada się, że już skonfigurowano Zerobus Ingest (endpoint, tabela docelowa i nazwa główna usługi). Zobacz Korzystanie z Zerobus Ingest. Informacje o tym, które pakiety SDK obsługują Avro, znajdziesz w repozytorium SDK Zerobus.

Zestaw SDK dla języka Python

Zainstaluj pakiet SDK z dodatkiem avro, który dołącza enkoder fastavro:

pip install "databricks-zerobus-ingest-sdk[avro]"

Przekaż schemat writer jako ciąg JSON do TableProperties za pomocą słowa kluczowego avro_schema, a następnie wczytuj rekordy jako słowniki:

import json

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

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 = {
    "type": "record",
    "name": "AirQuality",
    "fields": [
        {"name": "device_name", "type": "string"},
        {"name": "temp", "type": "int"},
        {"name": "humidity", "type": "long"},
    ],
}

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)
table_properties = TableProperties(TABLE_NAME, avro_schema=json.dumps(SCHEMA))
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)

try:
    for i in range(1000):
        stream.ingest_record_offset(
            {"device_name": f"sensor-{i}", "temp": 20 + (i % 5), "humidity": 55 + (i % 10)}
        )
    stream.flush()
finally:
    stream.close()

Aby przesłać dane, które już zakodowałeś, przekaż surowe bytes bezpośrednio do ingest_record_offset, bez typu wrappera. SDK przesyła je bez zmian. Bajty muszą być danymi Avro bez schematu, zakodowanymi przy użyciu tego samego schematu zapisującego. Tutaj są pokazane dosłownie, ale w praktyce pochodzą z systemu źródłowego:

# A schemaless Avro datum for {"device_name": "sensor-0", "temp": 21, "humidity": 55},
# encoded against the writer schema above: a length-prefixed "sensor-0", then temp
# and humidity as Avro varints.
payload = b"\x10sensor-0\x2a\x6e"
stream.ingest_record_offset(payload)

Rust SDK

Dodaj crate z feature avro, która wymaga Rust 1.88 lub nowszego:

cargo add databricks-zerobus-ingest-sdk --features avro
cargo add tokio --features macros,rt-multi-thread

Wybierz format Avro w .avro(schema_json) kreatorze strumieni, a następnie wprowadzaj AvroRecord wartości, które SDK koduje do schematu writera:

use databricks_zerobus_ingest_sdk::{AvroRecord, AvroValue, ZerobusSdk};

const SERVER_ENDPOINT: &str = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com";
const DATABRICKS_WORKSPACE_URL: &str = "https://dbc-a1b2c3d4-e5f6.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";

const AVRO_SCHEMA: &str = r#"{
  "type": "record",
  "name": "AirQuality",
  "fields": [
    {"name": "device_name", "type": "string"},
    {"name": "temp", "type": "int"},
    {"name": "humidity", "type": "long"}
  ]
}"#;

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

    let mut stream = sdk
        .stream_builder()
        .table(TABLE_NAME)
        .oauth(CLIENT_ID, CLIENT_SECRET)
        .avro(AVRO_SCHEMA)
        .build()
        .await?;

    for i in 0..1000 {
        let record = AvroValue::Record(vec![
            ("device_name".to_string(), AvroValue::String(format!("sensor-{i}"))),
            ("temp".to_string(), AvroValue::Int(20 + i % 5)),
            ("humidity".to_string(), AvroValue::Long((55 + i % 10) as i64)),
        ]);
        stream.ingest_record_offset(AvroRecord(record)).await?;
    }
    stream.flush().await?;
    stream.close().await?;

    Ok(())
}

Aby wysłać już zakodowane dane, przekaż AvroBytes(bytes) do ingest_record_offset zamiast AvroRecord.

Related

• Protokoły API: Protokoły API, przez które te formaty przechodzą.
• Zarządzanie schematami: Jak dokumenty są weryfikowane względem tabeli.
• Obsługiwane typy danych: typy danych Delta, Protobuf i Avro.