Nachrichtentypen

Wenn Sie mit den Zerobus Ingest SDKs über gRPC einlesen, wählen Sie, wie Datensätze auf einem bestimmten Stream codiert werden. Zerobus Ingest unterstützt vier Nachrichtenformate, sodass Sie je nach Arbeitslast zwischen Einfachheit, Typsicherheit und Durchsatz abwägen können: JSON, Protocol Buffers (protobuf), Apache Arrow und Avro (Beta). Jedes Format wird anhand Ihres Delta-Tabellenschemas validiert, bevor die Daten dauerhaft gemacht werden. Siehe Schemaverwaltung.

Das Diagramm zeigt JSON, Protubuf und Apache Arrow. Er schließt Avro nicht ein, das eine kompakte, zeilenorientierte Codierung ähnlich wie Protobuf verwendet.

Welches Format solltest du verwenden?

Die untenstehenden Ausschnitte zeigen die Form jedes Formats im Python SDK. Sie gehen davon aus, dass du den SDK-Client bereits erstellt hast und deine Ziel-Tabelle kennst. Für das vollständige Setup (Endpunkt, Tabelle, Service Principal) und Beispiele in jeder Sprache siehe Use Zerobus Ingest.

JSON

JSON ist der einfachste Weg: Sie senden Datensätze als JSON-Objekte ohne Schema-Definition zum Kompilieren. Es ist ideal für einen schnellen Einstieg, für die Prototypenerstellung und für Anwender, bei denen Benutzerfreundlichkeit wichtiger ist als reiner Durchsatz. Für Produktionslasten mit hohem Volumen ist ein Binärformat (Protobuf oder Arrow) effizienter.

Erstellen Sie einen Strom für JSON-Datensätze, indem Sie einen Tabellennamen ohne Deskriptor an TableProperties übergeben:

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})

Für den vollständigen JSON-Walkthrough siehe Write a Client.

Protokollpuffer

Protobuf bietet eine typsichere, kompakte binäre Codierung und ist das empfohlene Format für die meisten produktiven, zeilenorientierten Arbeitslasten. Du definierst ein Protobuf-Schema, das zu deiner Ziel-Delta-Tabelle passt (siehe Protobuf-Schema), kompilierst es, und das SDK speichert Datensätze Datensatz für Datensatz über gRPC.

Die Verwendung von Protobuf erfolgt in drei Schritten: Ein .proto Schema generieren, das zu deiner Tabelle passt, es in ein Sprachmodul kompilieren und dann Datensätze eintragen, indem du den Deskriptor an TablePropertiesweitergibst. Das folgende Beispiel verwendet das Python SDK.

1. Generiere ein .proto Schema aus deiner Tabelle. Das Python SDK enthält ein generate_proto Tool, das deine Delta-Tabelle liest und ein passendes Schema schreibt:

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"

Das generierte Schema verwendet die Syntax proto2, mit einem optionalen Feld für jede Delta-Spalte:

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

2. Kompiliere das Schema mit dem Protobuf-Compiler in ein Python-Modul:

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

Dadurch wird record_pb2.py generiert.

3. Eingaben von Datensätzen durch Weitergabe des kompilierten Deskriptors an TableProperties (der Standard für Protobuf). Das SDK verwendet den Deskriptor, um jeden Datensatz zu serialisieren:

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)

Das obige Beispiel verwendet das Python SDK. Das Tooling variiert je nach Sprache: Einige SDKs stellen ein generate_proto-Dienstprogramm bereit, das aus deiner Tabelle ein .proto generiert, während andere (wie Go und TypeScript) ein vorhandenes .proto kompilieren. Für die sprachspezifischen Schritte, siehe die Protobuf-Hinweise in den einzelnen SDK-Tabs in Write a client. Für Werkzeugquellen und vollständige Beispiele siehe das Zerobus SDK-Repository.

Apache-Pfeil

Die Apache Arrow-Aufnahme sendet Arrow-DatenRecordBatch direkt über dieselbe gRPC-Verbindung, anstatt jede Zeile zuerst in JSON oder Protobuf umzuwandeln. Es ist die beste Wahl, wenn Ihre Anwendung bereits Arrow-Daten erzeugt oder wenn Sie Zeilen in Chargen einnehmen, besonders bei breiten, numerisch lastigen oder analytisch orientierten Schemata, bei denen Zeilen-für-Zeilen-Serialisierung zusätzlichen Aufwand verursacht.

Arrow passt auch gut zu sehr großen Chargen. Im Gegensatz zu den JSON- und Protobuf-Batch-Methoden, die Alles-oder-Nichts sind und durch das Größenlimit pro Nachricht begrenzt sind, teilt der Arrow Flight-Pfad einen großen Batch in kleinere Transportnachrichten auf, die einzeln gesendet und bestätigt werden. Siehe Arrow-Flight-Batches bilden die Ausnahme und Verwenden Sie Arrow Flight mit Zerobus Ingest.

Öffnen Sie einen Arrow-Stream mit pyarrow.Schema und erfassen Sie RecordBatch-Daten:

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

stream.ingest_batch(batch)

Für den vollständigen Arrow Flight-Walkthrough, einschließlich Schema-Definition, Batching und Kompression, siehe Use Arrow Flight with Zerobus Ingest.

Avro

Apache Avro ist ein zeilenorientiertes, kompaktes, binäres, schemabasiertes Format. Wie bei Protobuf wird jeder Datensatz gegen ein Schema kodiert, anstatt seine Feldnamen inline zu tragen, was die Datensätze klein hält. Der Unterschied besteht darin, dass protobuf einen kompilierten .proto Deskriptor verwendet, während Avro seine Datensätze mit einem in JSON geschriebenen Schema beschreibt. Die Avro-Ingestion läuft über gRPC über die Zerobus-SDKs. Informationen dazu, welche SDKs Avro unterstützen, finden Sie im Zerobus SDK-Repository.

Avro passt gut, wenn dein Producer bereits mit Avro arbeitet, zum Beispiel wenn er Avro-Daten aus einem Upstream-System liest oder bereits über ein Avro-Writer-Schema verfügt.

Schema-Behandlung

Zerobus Ingest verwendet kein Schemaregister, und die Datensätze bei der Übertragung tragen keine Schema-ID oder keinen Fingerabdruck. Stattdessen lieferst du das Avro-Writer-Schema als JSON, wenn du den Stream erstellst, und jeder Datensatz in diesem Stream wird anhand dieses Schemas dekodiert. Das Schema wird von Anfang an gegen die Ziel-Delta-Tabelle validiert, sodass ein inkompatibles Schema abgelehnt wird, bevor Daten fließen.

Das Writer-Schema ist für die Lebensdauer des Streams festgelegt. Um mit einem anderen Schema zu erfassen, erstellen Sie einen neuen Stream.

Avro-Felder werden auf jeder Ebene mit Delta-Spalten nach Namen gepaart, daher spielt die Feldreihenfolge im Schreiberschema keine Rolle. Informationen dazu, wie Avro-Typen auf Delta-Spalten abgebildet werden und welche Spalten nicht als Ziel verwendet werden können, finden Sie unter Avro-Typen. Wie Datensätze gegen die Tabelle validiert werden, erfahren Sie unter Schema-Management.

Ein Avro-Writer-Schema muss jeden Typ inline deklarieren. Benannte Typreferenzen, die einen benannten record, enum, oder fixed Typ an anderer Stelle im Schema wiederverwenden, indem sie auf seinen Namen verweisen, werden nicht unterstützt.

Wie man Daten bereitstellt

Egal, ob Sie ein SDK verwenden oder nicht, Sie können einen Wert auf einmal oder in Batches erfassen. Was sich unterscheidet, ist, wer die Daten ins Avro-Format kodiert:

• Mit einem SDK können Sie entweder:
  • Lassen Sie das SDK Ihre Daten kodieren: Geben Sie dem SDK Ihre Avro-Datensätze, und es kodiert sie gegen das Schreibschema.
  • Vorcodierte Avro-Bytes senden: Wenn Sie bereits Avro-codierte Daten besitzen, z. B. aus einem Upstreamsystem, sendet das SDK diese unverändert.
• Ohne ein SDK sendet man beim direkten Aufruf der gRPC-API nur vorkodierte Avro-Bytes. Erstellen Sie den Stream mit dem Writer-Schema und senden Sie dann die rohe Avro-Binärcodierung Ihrer Daten, ohne Container-Datei-Header, Schema-Registry-ID-Präfix oder Fingerabdruck.

Schreibe einen Client

Die untenstehenden Beispiele verwenden die Python- und Rust-SDKs. Sie erstellen einen Stream, senden Datensätze, die das SDK gegen das Writer-Schema kodiert, und flushen. Sie gehen davon aus, dass Sie Zerobus Ingest bereits eingerichtet haben (Endpunkt, Zieltabelle und Service Principal). Siehe Use Zerobus Ingest. Welche SDKs Avro unterstützen, finden Sie im Zerobus SDK-Repository.

Python SDK

Installiere das SDK mit dem avro Zusatz, das den fastavro Encoder mitinstalliert:

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

Geben Sie das Writer-Schema als JSON-String mit dem Schlüsselwort TableProperties an avro_schema weiter und lesen Sie dann Datensätze als Wörterbücher ein:

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()

Um bereits kodierte Daten zu senden, übergib das rohe bytes direkt an ingest_record_offset, ohne Wrapper-Typ. Das SDK schickt sie unverändert durch. Die Bytes müssen ein schemaloses Avro-Datum sein, das unter Verwendung desselben Writer-Schemas kodiert ist. Sie werden hier als Literal dargestellt, aber in der Praxis stammen sie aus einem Upstream-System:

# 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

Füge die crate mit dem avro Feature hinzu, die Rust 1.88 oder später erfordert:

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

Wählen Sie das Avro-Format mit .avro(schema_json) im Stream-Generator aus, und erfassen Sie dann AvroRecord-Werte, die das SDK anhand des Writer-Schemas codiert:

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(())
}

Um einen Datensatz zu senden, den Sie bereits codiert haben, übergeben Sie AvroBytes(bytes) an ingest_record_offset anstelle von AvroRecord.

Related

• API-Protokolle: Die API-Protokolle, über die diese Formate übertragen werden.
• Schema-Management: Wie Datensätze an deiner Tabelle validiert werden.
• Unterstützte Datentypen: Unterstützte Delta-, Protobuf- und Avro-Datentypen.