Types de messages

Lorsque vous ingérez avec les SDK d’ingestion Zerobus via gRPC, vous choisissez comment les enregistrements sont encodés sur un flux donné. Zerobus Ingest prend en charge trois formats de message (JSON, Protocol Buffers (protobuf) et Apache Arrow), ce qui permet de faire un choix entre simplicité, sécurité des types et débit pour votre charge de travail. Chaque format est validé par rapport à votre schéma de table Delta avant que les données ne deviennent durables. Voir Gestion des schémas.

Quel format devriez-vous utiliser ?

Les extraits ci-dessous montrent la forme de chaque format dans le SDK Python. Ils supposent que vous avez déjà créé le client SDK et que vous connaissez votre table cible. Pour la configuration complète (terminaison, table, principal de service) et des exemples dans chaque langage, voir Utiliser l’ingestion du Zerobus.

JSON

Le JSON est la façon la plus simple de commencer : vous envoyez des enregistrements sous forme d’objets JSON sans définition de schéma à compiler. C’est idéal pour démarrer, prototyper et produire des produits où la commodité compte plus que le débit brut. Pour les charges de production à grand volume, un format binaire (protobuf ou Arrow) est plus efficace.

Créez un flux pour les enregistrements JSON en passant un nom de table à TableProperties sans descripteur :

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

Pour la solution complète en JSON, voir Écrire un client.

Mémoires tampon de protocole

Protobuf fournit un encodage binaire compact garantissant la sécurité des types, et constitue le format recommandé pour la plupart des charges de travail en production orientées lignes. Vous définissez un schéma protobuf qui s’adapte à votre table Delta cible (voir schéma Protobuf), vous le compilez, et le SDK ingère les enregistrements enregistrement par enregistrement via gRPC.

L’utilisation de protobuf se fait en trois étapes : générer un .proto schéma correspondant à votre table, le compiler dans un module de langage, puis ingérer les enregistrements en passant le descripteur à TableProperties. L’exemple suivant utilise le SDK Python.

1. Générez un .proto schéma à partir de votre table. Le SDK Python inclut un generate_proto outil qui lit votre table Delta et écrit un schéma correspondant :

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"

Le schéma généré utilise la syntaxe proto2, avec un champ optionnel pour chaque colonne Delta :

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

2. Compiler le schéma en module Python avec le compilateur protobuf :

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

Cela génère record_pb2.py.

3. Ingérer les enregistrements en passant le descripteur compilé à TableProperties (le code par défaut pour protobuf). Le SDK utilise le descripteur pour sérialiser chaque enregistrement :

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’exemple ci-dessus utilise le SDK Python. Les outils varient selon le langage : certains SDK livrent un generate_proto utilitaire qui génère un .proto a à partir de votre table, tandis que d’autres (comme Go et TypeScript) compilent un fichier existant .proto. Pour les étapes par langue, voir les notes protobuf dans chaque onglet SDK de Write a client. Pour les sources d’outils et des exemples complets, voir le dépôt Zerobus SDK.

Apache Arrow

L’ingestion Apache Arrow envoie les données ArrowRecordBatch directement via la même connexion gRPC, au lieu de convertir chaque ligne en JSON ou protobuf d’abord. C’est le meilleur choix lorsque votre application produit déjà des données Arrow ou lorsque vous ingérez des lignes par lots, surtout pour des schémas larges, très numériques ou orientés analytique où la sérialisation ligne par ligne ajoute de la surcharge.

Arrow convient aussi bien aux très grandes quantités. Contrairement aux méthodes JSON et protobuf batch, qui sont tout ou rien et limitées par la limite de taille par message, le trajet Arrow Flight divise un grand lot en messages de transport plus petits qui sont envoyés et accusés de réception individuellement. Voir Les lots Arrow Flight font exception et Utiliser Arrow Flight avec Zerobus Ingest.

Ouvrez un flux Arrow avec pyarrow.Schema et importez des données RecordBatch :

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

stream.ingest_batch(batch)

Pour le guide complet d’Arrow Flight, y compris la définition du schéma, le traitement par lots et la compression, consultez Utiliser Arrow Flight avec Zerobus Ingest.

Related

• Protocoles API : Les protocoles API sur lesquels ces formats circulent.
• Gestion de schéma : Comment les enregistrements sont validés par rapport à votre table.
• Types de données pris en charge : Types de données Delta et Protobuf pris en charge.