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 drei Nachrichtenformate (JSON, Protocol Buffers (Protobuf) und Apache Arrow), sodass Sie Einfachheit, Typsicherheit und Durchsatz für Ihre Arbeitslast eintauschen können. Jedes Format wird anhand Ihres Delta-Tabellenschemas validiert, bevor die Daten dauerhaft gemacht werden. Siehe Schemaverwaltung.

Dieselben 1.000 Datensätze in drei Nachrichtenformaten: protobuf als kompakte, typisierte Zeilenkodierung, Apache Arrow als ein spaltenorientierter Batch, dessen Metadaten und Puffer über den gesamten Batch amortisiert werden, und JSON als lesbarer Text mit wiederholten Feldnamen

Welches Format solltest du verwenden?

Format Am besten geeignet für: Hinweise
JSON Erste Schritte und einfache Producer. Die einfachste Option, ohne Schema-Definition zum Kompilieren. Praktisch, aber langsamer als die Binärformate für hochvolumige Arbeitslasten.
Protocol Buffers Produktions-, reihenorientierte, hochvolumige Ströme. Typsichere, kompakte binäre Codierung. Empfohlen für die meisten Produktionsarbeitslasten. Erfordert ein kompiliertes Schema.
Apache-Pfeil Spalten- oder batchorientierte Arbeitslasten. Sendet Apache Arrow Record Batches direkt und vermeidet dadurch eine zeilenweise Serialisierung. Am besten geeignet, wenn Ihre Daten bereits spaltenorientiert sind oder Sie sie batchweise importieren. In Beta. Siehe Verwendung von Arrow Flight mit Zerobus Ingest.

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

Important

Der Apache Arrow-Erfassung befindet sich in der Betaphase.

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.

  • Last updated on