Le tableau suivant liste les types de connecteurs Kafka Connect compatibles avec Managed Service pour Apache Kafka. Vous pouvez utiliser ces connecteurs pour intégrer Apache Kafka à vos applications et à d'autres services Google Cloud.
| Connecteur | Description | Cas d'utilisation |
|---|---|---|
| Récepteur BigQuery | Diffuse des données de sujets Kafka vers une table BigQuery. | Entreposage de données, analyse |
| Source Cloud SQL pour PostgreSQL | Lit les modifications au niveau des lignes d'une base de données Cloud SQL pour PostgreSQL et les écrit dans Kafka. | Intégration de bases de données, surveillance des événements de base de données et réponse à ces événements |
| Récepteur Cloud Storage | Transfère les données des sujets Kafka vers un bucket Cloud Storage. | Ingestion de lacs de données, archivage de données |
| Source PostgreSQL générique | Lit les modifications au niveau des lignes d'une base de données PostgreSQL et les écrit dans Kafka. | Intégration de bases de données, surveillance des événements de base de données et réponse à ces événements |
| Point de contrôle MirrorMaker 2.0 | Copie les décalages des clients d'un cluster Kafka vers un autre. | Réplication de données, reprise après sinistre, migration de données |
| Pulsation MirrorMaker 2.0 | Génère des messages de pulsation périodiques et les écrit dans un sujet | Surveiller l'état du cluster, la connectivité réseau ou la latence de réplication de bout en bout entre les clusters. |
| Source MirrorMaker 2.0 | Réplique les sujets et les données d'un cluster Kafka vers un autre cluster Kafka. | Réplication de données, reprise après sinistre, migration de données |
| Récepteur Pub/Sub | Diffuse les données des sujets Kafka vers un sujet Pub/Sub. | Intégration de services, notifications en temps réel |
| Source Pub/Sub | Diffuse des messages d'un abonnement Pub/Sub vers un sujet Kafka. | Ingestion de données en temps réel, architectures basées sur des événements |
Visiteurs ayant déjà réalisé une conversion
Les convertisseurs sont responsables de la sérialisation et de la désérialisation des données d'enregistrement Kafka. Ils traduisent le format d'octets bruts trouvé dans les sujets Kafka en une représentation de données interne et structurée utilisée par Kafka Connect.
Pour les connecteurs récepteurs, les convertisseurs désérialisent les données du format de transmission du sujet au format de données interne de Kafka Connect, que le connecteur utilise pour écrire dans le système cible.
Pour les connecteurs sources, les convertisseurs sérialisent les données du format de données interne de Kafka Connect au format de transmission spécifié pour le sujet Kafka.
Les convertisseurs s'assurent que le connecteur lit ou écrit les enregistrements Kafka dans un format compatible avec le système externe.
Lorsque vous configurez un connecteur, définissez les propriétés suivantes :
Convertisseur de clés (
key.converter) : convertisseur à utiliser lors de la sérialisation et de la désérialisation des clés d'enregistrement Kafka.Convertisseur de valeur (
value.converter) : convertisseur à utiliser lors de la sérialisation et de la désérialisation des valeurs d'enregistrement Kafka.
Si vous ne spécifiez pas de convertisseur, le type de convertisseur par défaut est org.apache.kafka.connect.converters.ByteArrayConverter, qui transmet les données au format d'octet brut.
Convertisseurs compatibles
Managed Service pour Apache Kafka est compatible avec les convertisseurs intégrés suivants :
| Visiteur ayant déjà réalisé une conversion | Format |
|---|---|
io.confluent.connect.avro.AvroConverter |
Apache Avro |
org.apache.kafka.connect.converters.BooleanConverter |
Booléen |
org.apache.kafka.connect.converters.ByteArrayConverter |
Tableau d'octets Type de convertisseur par défaut. Préserve le contenu exact des messages entre deux systèmes. |
org.apache.kafka.connect.converters.DoubleConverter |
Double |
org.apache.kafka.connect.converters.FloatConverter |
Float |
org.apache.kafka.connect.converters.IntegerConverter |
Integer |
org.apache.kafka.connect.json.JsonConverter |
JSON Pour les données JSON sans schéma, définissez également |
org.apache.kafka.connect.converters.LongConverter |
Long |
org.apache.kafka.connect.converters.ShortConverter |
Version courte |
org.apache.kafka.connect.storage.StringConverter |
Chaîne |
Le choix du convertisseur dépend du type de connecteur et des données que vous stockez dans Kafka. Pour en savoir plus, consultez la documentation du connecteur concerné.
Tâches
Un connecteur transfère des données en créant une ou plusieurs tâches qui s'exécutent en parallèle. Pour définir une limite supérieure au nombre de tâches créées par un connecteur, définissez la propriété de configuration tasks.max du connecteur. Le connecteur peut créer moins de tâches que cette valeur.
L'augmentation de la valeur de tasks.max peut améliorer le débit, mais aussi augmenter la consommation de ressources (processeur et mémoire). La valeur optimale dépend de la charge de travail et des ressources allouées aux nœuds de calcul de votre cluster Connect. Pour les connecteurs récepteurs, le nombre de partitions de sujets Kafka peut également affecter le parallélisme.
Règle de redémarrage des tâches
Vous pouvez définir la règle de redémarrage des tâches d'un connecteur, qui détermine le comportement en cas d'échec. Les connecteurs sont compatibles avec les règles suivantes :
Ne jamais redémarrer. Le connecteur ne redémarre pas les tâches ayant échoué. Cette règle correspond au comportement par défaut. Cela est utile pour le débogage ou dans les situations où une intervention manuelle est requise après une erreur.
Redémarrez avec un intervalle exponentiel entre les tentatives. Le connecteur redémarre une tâche ayant échoué après un certain délai (appelé période de backoff). Le délai augmente de manière exponentielle à chaque échec ultérieur. Cette règle est recommandée pour la plupart des charges de travail de production.
Si vous utilisez la règle d'intervalle exponentiel entre les tentatives, définissez également des valeurs pour l'intervalle minimal et maximal entre les tentatives. Le délai d'intervalle minimal doit être supérieur à 60 secondes et le délai d'intervalle maximal doit être inférieur à 7 200 secondes.
Transformations et prédicats
Managed Service pour Apache Kafka est compatible avec les transformations et les prédicats Kafka Connect par défaut.
Les transformations vous permettent de modifier des messages individuels avant qu'ils ne soient envoyés à Managed Service pour Apache Kafka (pour les connecteurs sources) ou au système externe (pour les connecteurs récepteurs). Vous pouvez utiliser une transformation pour masquer des données sensibles, ajouter des codes temporels ou renommer des champs.
Les prédicats vous permettent de filtrer les données en fonction de conditions spécifiques, en déterminant à quels messages une transformation s'applique en fonction des propriétés des messages.
Par exemple, pour configurer un connecteur récepteur de sorte qu'il ignore les messages contenant une clé d'en-tête DoNotProcess, ajoutez la configuration suivante :
transforms=dropMessage
transforms.dropMessage.type=org.apache.kafka.connect.transforms.Filter
transforms.dropMessage.predicate=hasKey
predicates=hasKey
predicates.hasKey.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.hasKey.name=DoNotProcess
Cette configuration effectue les opérations suivantes :
Configure un prédicat nommé
hasKeyde typeorg.apache.kafka.connect.transforms.predicates.HasHeaderKey. Ce prédicat correspond à tous les messages contenant un en-tête avec la cléDoNotProcess.Configure une transformation nommée
dropMessagede typeorg.apache.kafka.connect.transforms.Filter. Cette transformation supprime tous les messages qui correspondent au prédicat configuré.Associe la transformation au prédicat
hasKey. Cela garantit que seuls les messages contenant la clé d'en-têteDoNotProcesssont supprimés par la transformation.