Utilisez des API compatibles Kafka avec Zerobus Ingest
Aperçu
Cette fonctionnalité est en Préversion publique.
Zerobus Ingest propose des APIs de producteur compatibles avec Kafka qui vous permettent d'ingérer des données en utilisant n'importe quel client producteur Apache Kafka sans SDK Databricks. Vous pointez un producteur Kafka existant vers l'Endpoint Zerobus et produisez vers un sujet nommé d'après votre table cible, et les enregistrements sont stockés dans une table Delta Unity Catalog. Les APIs compatibles avec Kafka sont adaptées si vous disposez déjà d'un producteur Kafka, d'un collecteur utilisant le protocole Kafka ou d'outils émettant vers Kafka, et que vous souhaitez acheminer ces données vers Delta avec un minimum de modifications de code.
Les APIs compatibles Kafka s'exécutent sur TCP avec SASL_SSL et le mécanisme OAUTHBEARER et implémentent le sous-ensemble côté producteur du protocole Kafka, qui couvre Produce, Metadata, ApiVersions et les APIs de handshake SASL. Les APIs de consommateur, d'administration et transactionnelles ne sont pas disponibles. Les API sont en écriture seule.
Quand utiliser les APIs compatibles avec Kafka
Les APIs compatibles avec Kafka sont les plus adaptées dans les scénarios suivants :
- Vous souhaitez envoyer des données vers Delta sans adopter de SDK Zerobus, et vous exécutez déjà un producteur Kafka ou une application, un agent ou un collecteur qui émet vers Kafka.
- Vous souhaitez réutiliser votre configuration de producteur Kafka, votre traitement par batch et vos outils opérationnels existants.
- Vous envoyez des enregistrements JSON et n'avez pas besoin de Protocol Buffers (protobuf) ou d'Apache Arrow.
Si vous créez un nouveau client à partir de zéro et que vous souhaitez obtenir le throughput le plus élevé, des accusés de réception par enregistrement et une récupération automatique, utilisez un SDK Zerobus via gRPC plutôt que les APIs compatibles avec Kafka. Consultez Choisir une interface. Pour les workloads en colonnes ou par batch, consultez Utiliser Arrow Flight avec Zerobus Ingest.
Comment fonctionne le modèle d'ingestion
Les APIs compatibles Kafka mappent les concepts Kafka sur Zerobus Ingest comme suit :
- Sujets. Un nom de sujet Kafka est le nom complet de la table Unity Catalog à trois niveaux (
catalog.schema.table). La table cible doit déjà exister, car Zerobus ne crée jamais de sujets. - Enregistrements. Zerobus ingère uniquement la valeur de l'enregistrement, qui doit être un objet JSON encodé en UTF-8 correspondant au schéma de la table Delta cible. Il ignore les clés d'enregistrement, les en-têtes, ainsi que la partition et le Timestamp fournis par le client, et ne les conserve pas.
- Authentification. Chaque connexion s’authentifie avec un jeton OAuth Databricks limité à la table cible, présenté via SASL/
OAUTHBEARER. Voir Authentification. - Remerciements. Avec
acks=0, Zerobus ne renvoie aucune réponse. Le producteur ne peut donc pas confirmer l’ingestion. Avec n’importe quelle valeuracksdifférente de zéro, y comprisacks=1etacks=all, Zerobus ne renvoie une réponseProducequ’après avoir persisté durablement les enregistrements.
Zerobus est sans partition par conception. L’endpoint Zerobus est un broker logique unique et une partition unique, de sorte que chaque enregistrement pour un sujet est accusé de réception sur la partition 0. Les producteurs n’ont pas besoin d’en tenir compte. Zerobus Ingest monte en charge horizontalement pour gérer la charge entrante.
De plus, Zerobus Ingest fournit une remise au moins une fois. Les APIs compatibles Kafka sont en preview publique et ont leur propre quota ; elles partagent les caractéristiques de latence, de taille d’enregistrement et de table partitionnée du reste de Zerobus Ingest. Pour le quota Kafka et tous les autres quotas, consultez quotas Zerobus Ingest.
Authentification
Les APIs compatibles avec Kafka utilisent SASL/OAUTHBEARER. Le jeton porteur est un jeton OAuth Databricks que vous obtenez avec les identifiants client d’un Service Principal ayant accès à la table cible. Le jeton est limité à cette table via OAuth authorization_details et utilise la ressource zerobusDirectWriteApi, le même flux que l’ API REST Zerobus.
Les jetons OAuth expirent après une heure. Vous devez donc fournir le jeton via le rappel de fournisseur de jeton de votre client Kafka plutôt que sous la forme d'une chaîne statique. Le client récupère ensuite un nouveau jeton à chaque reconnexion. Les connexions ont également une durée de vie côté serveur d'environ 15 minutes, et Zerobus peut les fermer plus tôt pendant la maintenance du service. Quand Zerobus ferme une connexion, le producteur se reconnecte et se réauthentifie de lui-même. La réauthentification sur une connexion active n'est pas prise en charge.
Accordez au Service Principal les privilèges Unity Catalog requis sur la table cible avant de vous connecter. Consultez Créer un Service Principal et accorder des autorisations.
Écrire un client
L'exemple ci-dessous produit vers la même table air_quality utilisée dans les exemples Utiliser Zerobus Ingest. Il utilise kafka-python, mais tout client producteur Kafka prenant en charge SASL_SSL avec le mécanisme OAUTHBEARER fonctionne. Adaptez le modèle de fournisseur de jetons à votre bibliothèque cliente.
Le producteur se connecte au serveur bootstrap Zerobus sur le port 9092. Trouvez votre ID de workspace et votre région comme décrit dans Obtenir votre URL de workspace et votre endpoint Zerobus Ingest.
- Serveur bootstrap :
<workspace-id>.zerobus.<region>.gcp.databricks.com:9092
pip install kafka-python requests
Étape 1 : Créer un fournisseur de jetons
Zerobus authentifie chaque connexion avec un jeton OAuth Databricks éphémère et limité à la table. Comme le jeton expire, transmettez un rappel qui génère un nouveau jeton à la demande plutôt qu'un jeton statique.
La fonction fetch_zerobus_token() échange vos identifiants de Service Principal contre un jeton limité à la table cible, et ZerobusTokenProvider l’encapsule dans l’interface de rappel attendue par kafka-python.
import json
import requests
from kafka.sasl.oauth import AbstractTokenProvider
# See "Get your workspace URL and Zerobus Ingest endpoint" in zerobus-ingest.md.
WORKSPACE_ID = "1234567890123456"
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"
def fetch_zerobus_token():
catalog, schema, table = TABLE_NAME.split(".")
authorization_details = [
{
"type": "unity_catalog_privileges",
"privileges": ["USE CATALOG"],
"object_type": "CATALOG",
"object_full_path": catalog,
},
{
"type": "unity_catalog_privileges",
"privileges": ["USE SCHEMA"],
"object_type": "SCHEMA",
"object_full_path": f"{catalog}.{schema}",
},
{
"type": "unity_catalog_privileges",
"privileges": ["SELECT", "MODIFY"],
"object_type": "TABLE",
"object_full_path": TABLE_NAME,
},
]
response = requests.post(
f"{WORKSPACE_URL}/oidc/v1/token",
auth=(CLIENT_ID, CLIENT_SECRET),
data={
"grant_type": "client_credentials",
"scope": "all-apis",
"resource": f"api://databricks/workspaces/{WORKSPACE_ID}/zerobusDirectWriteApi",
"authorization_details": json.dumps(authorization_details),
},
timeout=30,
)
response.raise_for_status()
return response.json()["access_token"]
# kafka-python calls token() whenever it needs a fresh OAuth token.
class ZerobusTokenProvider(AbstractTokenProvider):
def token(self):
return fetch_zerobus_token()
Étape 2 : Configurer le producteur et envoyer les enregistrements
Pointez le producteur vers le serveur bootstrap, configurez SASL_SSL avec le mécanisme OAUTHBEARER et transmettez le fournisseur de jetons de l’étape 1. Utilisez acks="all" afin que chaque batch soit confirmé uniquement après avoir été durablement persisté, et envoyez les enregistrements sans compression.
from kafka import KafkaProducer
BOOTSTRAP_SERVERS = "1234567890123456.zerobus.us-west-2.cloud.databricks.com:9092"
producer = KafkaProducer(
bootstrap_servers=BOOTSTRAP_SERVERS,
security_protocol="SASL_SSL",
sasl_mechanism="OAUTHBEARER",
sasl_oauth_token_provider=ZerobusTokenProvider(),
# Wait for durable acknowledgment before treating a record as ingested.
acks="all",
# Compression is not supported by the endpoint; send records uncompressed.
compression_type=None,
)
# Each send() returns a future immediately. The topic is the full table name.
futures = [
producer.send(
topic=TABLE_NAME,
value=json.dumps(
{"device_name": f"sensor-{i}", "temp": 20 + i % 15, "humidity": 50 + i % 40}
).encode("utf-8"),
)
for i in range(1000)
]
producer.flush()
# Block on each future to confirm every record was durably acknowledged.
for future in futures:
future.get(timeout=30)
producer.close()
print("All records ingested successfully")
Options de configuration
Les APIs compatibles Kafka implémentent le sous-ensemble producteur du protocole Kafka. Configurez votre producteur selon les options suivantes.
Option | Détails |
|---|---|
Format d’enregistrement | JSON uniquement. Chaque valeur d'enregistrement doit être un objet JSON encodé en UTF-8 qui correspond au schéma de la table cible. Pour envoyer des données au format Protobuf ou Avro, utilisez un SDK Zerobus. |
Compression | Non pris en charge. Envoyez des batchs non compressés, par exemple |
Champs d'enregistrement | Valeur uniquement. Zerobus ingère la valeur de l’enregistrement et ne persiste pas les clés, les en-têtes, les affectations de partition ou les Timestamp. Les enregistrements avec une valeur nulle (tombstones) sont ignorés. |
Taille de la demande | Jusqu'à 100 Mo par requête. Zerobus ferme la connexion sans réponse lorsqu'une requête dépasse cette limite. Chaque enregistrement est également limité à 10 Mo. |
Prise en charge des API | Écriture seule. Zerobus accepte les requêtes |
Private Link front-end | Non pris en charge. Connectez-vous plutôt via l’endpoint public. |
Schéma | Appliqué. Zerobus rejette les enregistrements dont les champs ne correspondent pas au schéma de la table cible et traite les colonnes Delta supplémentaires autorisant les valeurs NULL comme une modification non bloquante. Pour capturer les champs non correspondants au lieu de les rejeter, configurez une colonne de secours. |
Pour les limites de latence, de quota, de taille d'enregistrement et de table partitionnée, consultez les quotas de Zerobus Ingest.
Bonnes pratiques
Suivez ces directives pour obtenir les meilleures performances et la meilleure fiabilité avec les APIs compatibles avec Kafka.
- Réutilisez un producteur à longue durée de vie pour de nombreux enregistrements plutôt que d’en créer un par batch, car la création du producteur et la négociation SASL entraînent des coûts de configuration.
- Laissez le producteur accumuler les enregistrements dans des batchs, par exemple en ajustant
linger.msetbatch.size, au lieu d'effectuer un vidage après chaque enregistrement. Le traitement par batch est le levier le plus important pour le throughput. - Utilisez
acks=allpour obtenir un accusé de réception durable pour chaque batch, en respectant la sémantique « at-least-once » de Zerobus. - Récupérez les jetons OAuth via le rappel du fournisseur de jetons de votre client afin qu'ils refresh automatiquement lors de la reconnexion, plutôt que de transmettre un jeton statique qui expire.
- Exécutez le producteur dans la même région cloud que l'endpoint Zerobus pour un throughput maximal.
Gestion des erreurs
Zerobus signale les échecs en utilisant les codes d’erreur Kafka standard sur le sujet et la partition concernés. Sur Produce v8 et versions ultérieures, chaque partition défaillante comporte également un message d’erreur qui décrit la cause. Les codes sont :
Erreur Kafka | Signification |
|---|---|
| Le jeton OAuth est manquant ou invalide, la connexion n'est pas authentifiée ou l'aperçu de l' Endpoint Kafka Zerobus Ingest n'est pas activé pour le workspace. |
| Le jeton ne couvre pas la table ou le service principal ne dispose pas d'un privilège Unity Catalog requis dessus. |
| La table cible n’existe pas ou a été supprimée. |
| Le sujet n’est pas un nom de table à trois niveaux valide, ou le type ou le format de la table ne prend pas en charge l’ingestion Zerobus, par exemple une vue. |
| Un enregistrement a échoué à la validation du schéma ou n'a pas pu être décodé en tant que JSON UTF-8. |
| Un seul enregistrement dépasse la limite de taille d’enregistrement de 10 Mo. Voir Taille de l’enregistrement. |
| Le batch a été compressé. Envoyez les enregistrements non compressés. |
| Le workspace est empêché d'ingérer, par exemple parce qu'il a dépassé son quota. |
| Le service est temporairement soumis à une charge importante. L’erreur est renouvelable et les producteurs Kafka la renouvellent automatiquement. |
| Le client a tenté de se réauthentifier sur une connexion authentifiée. La réauthentification n’est pas prise en charge. |
| Le client a envoyé une requête |
| Une erreur interne s’est produite. Nouvelle tentative de la requête. |
Après une requête Produce échouée ou une authentification échouée, Zerobus renvoie les codes d’erreur et ferme la connexion. Si le producteur utilise acks=0, Zerobus ferme la connexion sans réponse. Les producteurs Kafka se reconnectent automatiquement, mais concevez votre client pour faire remonter les échecs d’envoi, par exemple en inspectant le résultat de chaque envoi, afin que les enregistrements ne soient pas supprimés silencieusement.
Ressources supplémentaires
- Utiliser Zerobus Ingest: si vous n’avez pas encore configuré Zerobus Ingest, start ici pour obtenir des instructions sur la recherche de l’URL de votre Workspace, la création de la table Delta cible et la configuration d’un Service Principal. Ces étapes sont partagées entre toutes les interfaces.
- Quotas de Zerobus Ingest: examinez les quotas default de Zerobus avant tout déploiement en production.
- Utiliser Arrow Flight avec Zerobus Ingest: pour une ingestion en colonnes ou par batch via gRPC.