Tipi di messaggio

Quando ingeri con gli SDK di Zerobus Ingest tramite gRPC, scegli come vengono codificati i record su un dato stream. Zerobus Ingest supporta quattro formati di messaggio, quindi puoi bilanciare semplicità, sicurezza dei tipi e throughput per il tuo carico di lavoro: JSON, Protocol Buffer (protobuf), Apache Arrow e Avro (Beta). Ogni formato viene validato rispetto allo schema della tabella Delta prima che i dati diventino resi durevoli. Vedi Gestione degli schemi.

Il diagramma mostra JSON, protobuf e Apache Arrow. Non comprende Avro, che utilizza una codifica compatta orientata alle righe simile a protobuf.

Quale formato dovresti usare?

Gli estratti qui sotto mostrano la forma di ciascun formato nell'SDK Python. Presuppongono che tu abbia già creato il client SDK e conosca la tabella di destinazione. Per la configurazione completa (endpoint, tabella, principale di servizio) e esempi in ogni linguaggio, vedi Usa Zerobus Ingest.

JSON

JSON è il modo più semplice per iniziare: invii i record come oggetti JSON senza definizione di schema da compilare. È ideale per avviare e funzionare, prototipare e produrre dove la comodità conta più della capacità bruta. Per carichi di lavoro di produzione ad alto volume, un formato binario (protobuf o Arrow) è più efficiente.

Crea un flusso per i record JSON passando il nome di una tabella a TableProperties senza descrittore:

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

Per una guida dettagliata completa su JSON, vedi Scrivere un client.

Buffer di protocollo

Protobuf fornisce una codifica binaria compatta con controllo dei tipi ed è il formato consigliato per la maggior parte dei workload di produzione basati su righe. Definisci uno schema protobuf compatibile con la tua tabella Delta di destinazione (vedi schema Protobuf), lo compili e l'SDK acquisisce i record uno alla volta tramite gRPC.

L'uso di protobuf richiede tre passaggi: generare uno .proto schema che corrisponda alla tua tabella, compilarlo in un modulo linguistico, poi ingerire i record passando il descrittore a TableProperties. Il seguente esempio utilizza l'SDK Python.

1. Genera uno .proto schema dalla tua tabella. L'SDK Python include uno generate_proto strumento che legge la tua tabella Delta e scrive uno schema corrispondente:

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"

Lo schema generato utilizza proto2 la sintassi, con un campo opzionale per ogni colonna Delta:

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

2. Compilare lo schema in un modulo Python con il compilatore protobuf:

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

In questo modo viene generato record_pb2.py.

3. Ingerire i record passando il descrittore compilato a TableProperties (il valore predefinito per protobuf). L'SDK usa il descrittore per serializzare ogni record:

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)

L'esempio sopra utilizza l'SDK Python. Gli strumenti variano a seconda del linguaggio: alcuni SDK forniscono un'utilità generate_proto che genera a .proto dalla tua tabella, mentre altri (come Go e TypeScript) compilano un .proto. Per i passaggi per linguaggio, consulta le note protobuf in ogni scheda SDK di Scrivi un client. Per le fonti degli strumenti ed esempi completi, vedi il repository Zerobus SDK.

Apache Arrow

L'ingestione di Apache Arrow invia i dati ArrowRecordBatch direttamente sulla stessa connessione gRPC, invece di convertire prima ogni riga in JSON o protobuf. È la scelta migliore quando la tua applicazione produce già dati Arrow o quando acquisisci righe in batch, soprattutto per schemi estesi, a prevalenza numerica o orientati all'analisi, in cui la serializzazione riga per riga aggiunge overhead.

Arrow è anche una buona scelta per lotti molto grandi. A differenza dei metodi batch JSON e protobuf, che sono tutto o niente e limitati dal limite di dimensione per messaggio, il percorso Arrow Flight divide un grande batch in messaggi di trasporto più piccoli che vengono inviati e riconosciuti individualmente. Consulta I batch Arrow Flight fanno eccezione e Usa Arrow Flight con Zerobus Ingest.

Apri un flusso Arrow con un pyarrow.Schema e acquisisci i dati RecordBatch:

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

stream.ingest_batch(batch)

Per la guida completa di Arrow Flight, inclusa la definizione dello schema, l'elaborazione in batch e la compressione, consulta Usa Arrow Flight con Zerobus Ingest.

Avro

Apache Avro è un formato compatto, orientato alle righe, binario e guidato da schema. Come protobuf, ogni record viene codificato su uno schema invece di portare i nomi dei campi in linea, il che mantiene i record piccoli. La differenza è che protobuf utilizza un descrittore compilato .proto , mentre Avro descrive i suoi record con uno schema scritto in JSON. L'ingestione Avro viene eseguita tramite gRPC tramite gli SDK Zerobus. Per sapere quali SDK supportano Avro, vedi il repository Zerobus SDK.

Avro è adatto quando il tuo produttore lavora già con Avro, ad esempio quando legge dati Avro da un sistema upstream o possiede già uno schema writer Avro.

Gestione dello schema

Zerobus Ingest non utilizza un registro degli schemi e i record in transito non riportano alcun ID dello schema né impronta dello schema. Invece, fornisci lo schema dello scrittore Avro come JSON quando crei il flusso, e ogni record su quel flusso viene decodificato rispetto a quello schema. Lo schema viene convalidato rispetto alla tabella Delta di destinazione fin dall'inizio, quindi uno schema incompatibile viene rifiutato prima che i dati vengano trasferiti.

Lo schema del writer è fisso per tutta la durata del flusso. Per ingerire con uno schema diverso, crea un nuovo stream.

I campi Avro sono abbinati alle colonne Delta per nome a ogni livello, quindi l'ordine dei campi nello schema dello scrittore non conta. Per capire come i tipi Avro si mappano alle colonne Delta e quali colonne non possono essere mirate, vedi tipi Avro. Per informazioni su come vengono convalidati i record rispetto alla tabella, vedi Gestione degli schemi.

Uno schema di scrittore Avro deve dichiarare ogni tipo inline. I riferimenti a tipo nominato, che riutilizzano un tipo denominato record, enum o fixed altrove nello schema facendo riferimento al suo nome, non sono supportati.

Come fornire i dati

Che tu usi o meno un SDK, puoi acquisire un valore alla volta o in batch. Ciò che differisce è chi codifica i dati in Avro:

• Con un SDK, puoi scegliere tra:
  • Lascia che l'SDK codifichi i tuoi dati: consegna all’SDK i tuoi record Avro e li codifica in base allo schema del writer.
  • Invia byte Avro pre-codificati: se già possiedi dati codificati Avro, ad esempio da un sistema a monte, l'SDK li invia così come sono.
• Senza un SDK, chiamando direttamente l'API gRPC, invii solo byte Avro pre-codificati. Crea il flusso con lo schema dello scrittore, poi invia la codifica binaria Avro grezza dei tuoi dati, senza intestazione di file container, prefisso ID registro schema o impronta digitale.

Scrivere un client

Gli esempi seguenti utilizzano gli SDK di Python e Rust. Creano un flusso, inviano i record che l'SDK codifica rispetto allo schema dello scrittore ed eseguono il flush. Presumono che tu abbia già configurato Zerobus Ingest (endpoint, target table e service principal). Vedi Usa Zerobus Ingest. Per sapere quali SDK supportano Avro, vedi il repository Zerobus SDK.

PYTHON SDK

Installa l'SDK con l'extra avro , che richiama l'encoder fastavro :

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

Passa lo schema writer come stringa JSON a TableProperties con la parola chiave avro_schema, poi acquisisci i record come dizionari:

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

Per inviare i dati che hai già codificato, passa il raw ingest_record_offset direttamente a bytes, senza tipo wrapper. L’SDK li invia senza cambiamenti. I byte devono essere un dato Avro senza schema codificato rispetto allo stesso schema writer. Qui sono mostrati come valore letterale, ma in pratica provengono da un sistema a monte:

# 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

Aggiungi il crate con la funzionalità avro, che richiede Rust 1.88 o successiva:

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

Seleziona il formato Avro con .avro(schema_json) sul builder di stream, poi acquisisci i valori AvroRecord che l'SDK codifica rispetto allo schema dello scrittore:

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

Per inviare un dato già codificato, passa AvroBytes(bytes) a ingest_record_offset invece di AvroRecord.

Related

• Protocolli API: I protocolli API su cui questi formati viaggiano.
• Gestione dello schema: come i record vengono convalidati rispetto alla tua tabella.
• Tipi di dati supportati: Supporta i tipi di dati Delta, Protobuf e Avro.