How to create a Tablestore sink connector to export data from Kafka - ApsaraMQ for Kafka

Prérequis

ApsaraMQ for Kafka
- La fonctionnalité de connecteur est activée pour l'instance. Pour plus d'informations, consultez Activer les connecteurs.
- Un topic source a été créé pour l'instance. Pour plus d'informations, consultez Étape 1 : Créer un topic.
Tablestore
- Tablestore est activé et une instance a été créée. Pour plus d'informations, consultez Activer Tablestore et créer une instance.

Remarques

L'exportation des données d'un topic source d'une instance ApsaraMQ for Kafka vers Tablestore s'effectue uniquement au sein de la même région. Pour connaître les limites des connecteurs, reportez-vous à Limites d'utilisation.
Lors de la création d'un connecteur, ApsaraMQ for Kafka génère automatiquement un rôle lié au service.
- Si aucun rôle lié au service n'existe, ApsaraMQ for Kafka en crée un automatiquement, ce qui permet à ApsaraMQ for Kafka d'exporter des données vers Tablestore.
- Si un rôle lié au service existe déjà, ApsaraMQ for Kafka n'en crée pas de nouveau.
Pour plus d'informations sur les rôles liés au service, consultez Rôles liés au service.

Procédure

Cette rubrique détaille l'utilisation d'un connecteur sink Tablestore pour exporter des données d'un topic source d'une instance ApsaraMQ for Kafka vers Tablestore.

Facultatif : Créez les topics et le group requis pour le connecteur sink Tablestore.

Si vous n'avez pas besoin de personnaliser les topics et le group, ignorez cette étape et sélectionnez Auto à l'étape suivante.

Important
Certains topics requis par le connecteur sink Tablestore doivent utiliser le moteur de stockage Local. Pour les instances ApsaraMQ for Kafka en version majeure 0.10.2, la création manuelle de topics utilisant le moteur de stockage Local est impossible. Ces topics ne peuvent être créés qu'automatiquement.
1. Créer les topics pour le connecteur sink Tablestore
2. Créer le group pour le connecteur sink Tablestore
Créer et déployer le connecteur sink Tablestore
Vérifiez le résultat
1. Envoyer un message de test
2. Consulter les données dans la table

Créer les topics pour le connecteur sink Tablestore

Dans la console ApsaraMQ for Kafka, créez manuellement les cinq topics nécessaires au connecteur sink Tablestore : un topic d'offset de tâche, un topic de configuration de tâche, un topic d'état de tâche, un topic de file d'attente de lettres mortes et un topic de données d'erreur. Le nombre de partitions requis et le moteur de stockage varient selon le topic. Pour plus d'informations, consultez Liste des paramètres du service source.

Connectez-vous à la console ApsaraMQ for Kafka.
Sur la page Overview, sélectionnez une région dans la section Resource Distribution.

Important
Les topics doivent être créés dans la même région que votre application, c'est-à-dire la région où l'instance ECS est déployée. L'utilisation de topics entre régions n'est pas prise en charge. Par exemple, si un topic est créé dans la région Chine (Pékin), le producteur et le consommateur de messages doivent également s'exécuter sur une instance ECS située dans la région Chine (Pékin).
Sur la page Instances, cliquez sur le nom de l'instance cible.
Dans le volet de navigation de gauche, cliquez sur Topics.
Sur la page Topics, cliquez sur Create Topic.

Dans le panneau Create Topic, définissez les propriétés du topic, puis cliquez sur OK.

Paramètre	Description	Exemple
Name	Nom du topic. Remarque Dans Kafka, les noms de topic tels que `xxx_xxx` et `xxx.xxx` sont considérés comme identiques. Une erreur se produit si vous tentez de créer un topic avec un nom dupliqué.	demo
Description	Brève description du topic.	demo test
Partitions	Nombre de partitions du topic.	12
Storage Engine Remarque Actuellement, la sélection d'un type de moteur de stockage n'est possible que pour les instances non Serverless Professional Edition. Pour les autres instances, ce paramètre n'est pas pris en charge et prend la valeur Cloud Storage par défaut.	Moteur de stockage des messages du topic. ApsaraMQ for Kafka prend en charge les deux moteurs de stockage suivants. Cloud Storage : La couche sous-jacente utilise des disques Alibaba Cloud. Ce moteur de stockage offre une faible latence, des performances élevées, une grande durabilité et une haute fiabilité. Il repose sur un mécanisme distribué à trois réplicas. Si l'Instance Edition de l'instance est Standard (High Write), le moteur de stockage doit obligatoirement être Cloud Storage. Local Storage : Utilise l'algorithme natif de réplication ISR (in-sync replica) de Kafka et un mécanisme distribué à trois réplicas.	Cloud Storage
Message Type	Type des messages du topic. Normal Message : Par défaut, les messages partageant la même clé sont distribués dans la même partition, et les messages d'une partition sont stockés selon leur ordre d'envoi. En cas de panne d'une machine du cluster, l'ordre des messages peut être perturbé. Lorsque le paramètre Storage Engine est défini sur Cloud Storage, l'option Normal Message est sélectionnée par défaut. Partitionally Ordered Message : Par défaut, les messages partageant la même clé sont distribués dans la même partition, et les messages d'une partition sont stockés selon leur ordre d'envoi. En cas de panne d'une machine du cluster, l'ordre des messages reste garanti au sein de la partition. Toutefois, l'envoi de messages vers certaines partitions peut échouer. Les opérations normales reprennent après la récupération des partitions. Lorsque le paramètre Storage Engine est défini sur Local Storage, l'option Partitionally Ordered Message est sélectionnée par défaut.	Normal Message
Log Cleanup Policy	Politique de nettoyage des journaux du topic. Lorsque le paramètre Storage Engine est défini sur Local Storage, vous devez configurer la Log Cleanup Policy. La définition du moteur de stockage sur Local Storage n'est possible que pour les instances Professional Edition. Les instances Standard Edition ne sont pas prises en charge. ApsaraMQ for Kafka prend en charge les deux politiques de nettoyage suivantes. Delete : Politique de nettoyage des messages par défaut. Si la capacité du disque est suffisante, les messages sont conservés pendant la durée de rétention maximale. Si la capacité du disque est insuffisante (généralement lorsque l'utilisation du disque dépasse 85 %), les anciens messages sont supprimés prématurément afin de garantir la disponibilité du service. Compact : Utilise la politique de nettoyage Log Compaction de Kafka. La politique Log Compaction garantit que pour les messages partageant la même clé, la valeur la plus récente est toujours conservée. Cela s'applique principalement aux scénarios tels que la récupération d'état après une panne système ou le rechargement du cache après un redémarrage système. Par exemple, lors de l'utilisation de Kafka Connect ou de Confluent Schema Registry, vous devez utiliser un Kafka Compact Topic pour stocker l'état du système ou les informations de configuration. Important Les topics compacts sont généralement utilisés uniquement dans certains composants de l'écosystème, tels que Kafka Connect ou Confluent Schema Registry. Ne définissez pas cette propriété pour les topics dans d'autres scénarios d'envoi et de réception de messages. Pour plus d'informations, consultez la bibliothèque de démos ApsaraMQ for Kafka.	Compact
Tag	Tags du topic.	demo

Une fois le topic créé, il apparaît dans la liste de la page Topics.

Créer un group pour le connecteur sink Tablestore

Vous pouvez créer manuellement un group pour la tâche de synchronisation des données du connecteur sink Tablestore dans la console ApsaraMQ for Kafka. Le nom de ce group doit respecter le format connect-Nom de la tâche. Pour plus d'informations, consultez Paramètres du service source.

Connectez-vous à la console ApsaraMQ for Kafka.
Sur la page Overview, sélectionnez une région dans la section Resource Distribution.
Sur la page Instances, cliquez sur le nom de l'instance cible.
Dans le volet de navigation de gauche, cliquez sur Groups.
Sur la page Groups, cliquez sur Create Group.
Dans le panneau Create Group, saisissez un nom pour le group dans la zone de texte Group ID, entrez une brève description dans la zone de texte Description, ajoutez des tags au group, puis cliquez sur OK.

Une fois le group créé, il apparaît dans la liste de la page Groups.

Créer et déployer un connecteur sink Tablestore

Créez et déployez un connecteur sink Tablestore pour synchroniser des données d'une instance ApsaraMQ for Kafka vers une table Tablestore.

Connectez-vous à la console ApsaraMQ for Kafka.
Sur la page Overview, sélectionnez une région dans la section Resource Distribution.
Dans le volet de navigation de gauche, cliquez sur Connectors.
Sur la page Connectors, sélectionnez l'instance à laquelle appartient le connecteur dans la liste déroulante Select Instance, puis cliquez sur Create Connector.

Dans l'assistant Create Connector, suivez les étapes ci-dessous.

Sous l'onglet Configure Basic Information, configurez les paramètres suivants, puis cliquez sur Next.

Paramètre	Description	Exemple
Name	Nom du connecteur. Le nom doit respecter les exigences suivantes : Le nom peut comporter jusqu'à 48 caractères et contenir uniquement des chiffres, des lettres minuscules et des traits d'union (-). Il ne peut pas commencer par un trait d'union (-). Le nom doit être unique au sein d'une instance ApsaraMQ for Kafka. La tâche de synchronisation des données du connecteur utilise un consumer group nommé `connect-task-name`. Si vous ne créez pas ce consumer group manuellement, le système le crée automatiquement.	kafka-ts-sink
Instance	Par défaut, le nom de l'instance et son ID sont affichés.	demo alikafka_post-cn-st21p8vj****

Sous l'onglet Configure Source Service, définissez Data Source sur ApsaraMQ for Kafka, configurez les paramètres suivants, puis cliquez sur Next.

Remarque

Si vous avez déjà créé le topic et le group requis, sélectionnez Manual pour la création des ressources et saisissez leurs informations. Sinon, sélectionnez Auto.

Tableau 1. Paramètres du service source

Paramètre	Description	Exemple
Data Source Topic	Topic source pour la synchronisation des données.	ts-test-input
Consumer Thread Concurrency	Nombre de threads consommateurs simultanés pour le topic source de données. La valeur par défaut est 6. Valeurs valides : 1 2 3 6 12	6
Consumer Offset	Position de départ pour la consommation des messages. Valeurs valides : Earliest Offset : Consomme les messages à partir de l'offset le plus ancien. Latest Offset : Consomme les messages à partir de l'offset le plus récent.	Earliest Offset
VPC ID	ID du Virtual Private Cloud (VPC) dans lequel s'exécute la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Par défaut, ce paramètre correspond au VPC de l'instance ApsaraMQ for Kafka et ne nécessite aucune configuration.	vpc-bp1xpdnd3l***
vSwitch ID	ID du vSwitch dans lequel s'exécute la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Le vSwitch doit se trouver dans le même VPC que l'instance ApsaraMQ for Kafka. Par défaut, ce paramètre correspond au vSwitch spécifié lors du déploiement de l'instance ApsaraMQ for Kafka.	vsw-bp1d2jgg81***
Failure Handling Policy	Politique de gestion des échecs d'envoi de messages dans une partition de topic. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Valeurs valides : Continue Subscription : Maintient l'abonnement à la partition de topic où l'erreur s'est produite et imprime les journaux d'erreurs. Stop Subscription : Interrompt l'abonnement à la partition de topic où l'erreur s'est produite et imprime les journaux d'erreurs. Remarque Pour plus d'informations sur la consultation des journaux, voir Gérer les connecteurs. Pour savoir comment trouver des solutions en fonction des codes d'erreur, voir Codes d'erreur.	Continue Subscription
Resource Creation Method	Méthode de création des topics et du group requis. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Auto Manual	Auto
Connector Consumer Group	group consommateur utilisé par la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Le nom du group doit respecter le format connect-task-name.	connect-cluster-kafka-ots-sink
Task Offset Topic	Topic stockant les offsets consommateurs. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Topic : Le nom du topic doit commencer par connect-offset. Partitions : Le nombre de partitions du topic doit être supérieur à 1. Moteur de stockage : Le moteur de stockage du topic doit être Local Storage. `cleanup.policy` : La politique de nettoyage des journaux du topic doit être `compact`.	connect-offset-kafka-ots-sink
Task Configuration Topic	Topic stockant les configurations de tâche. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Topic : Le nom du topic doit commencer par connect-config. Partitions : Le nombre de partitions du topic doit être égal à 1. Moteur de stockage : Le moteur de stockage du topic doit être Local Storage. `cleanup.policy` : La politique de nettoyage des journaux du topic doit être `compact`.	connect-config-kafka-ots-sink
Task Status Topic	Topic stockant l'état des tâches. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Topic : Le nom du topic doit commencer par connect-status. Partitions : Le nombre recommandé de partitions est 6. Moteur de stockage : Le moteur de stockage du topic doit être Local Storage. `cleanup.policy` : La politique de nettoyage des journaux du topic doit être `compact`.	connect-status-kafka-ots-sink
Dead-letter Queue Topic	Topic utilisé pour stocker les données d'erreur du framework Kafka Connect. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Afin d'économiser les ressources de topic, vous pouvez utiliser le même topic pour la dead-letter queue et le Error Data Topic. Topic : Le nom du topic doit commencer par connect-error. Partitions : Le nombre recommandé de partitions est 6. Moteur de stockage : Le moteur de stockage du topic peut être Local Storage ou Cloud Storage.	connect-error-kafka-ots-sink
Error Data Topic	Topic utilisé pour stocker les données d'erreur du sink. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Afin d'économiser les ressources de topic, vous pouvez utiliser le même topic pour le Dead-letter Queue Topic et le topic de données d'erreur. Topic : Le nom du topic doit commencer par connect-error. Partitions : Le nombre recommandé de partitions est 6. Moteur de stockage : Le moteur de stockage du topic peut être Local Storage ou Cloud Storage.	connect-error-kafka-ots-sink

Sous l'onglet Configure Destination Service, définissez Destination Service sur Tablestore, configurez les paramètres suivants, puis cliquez sur Create.

Paramètre	Description	Exemple
Instance Name	Nom de l'instance Tablestore.	k00eny67****
Automatically Create Destination Table	Indique s'il faut créer automatiquement une table dans Tablestore. Yes : Une table est automatiquement créée dans Tablestore pour stocker les données synchronisées, en fonction du nom de table configuré. No : Une table existante est utilisée pour stocker les données synchronisées.	Yes
Destination Table Name	Nom de la table stockant les données synchronisées. Si vous définissez Automatically Create Destination Table sur No, le nom de la table doit correspondre au nom d'une table existante dans l'instance Tablestore.	kafka_table
Tablestore	Type de la table stockant les données synchronisées. Wide Column Model TimeSeries Model	Wide Column Model
Message Key Format	Format d'entrée de la clé du message. Les valeurs valides sont String et JSON. La valeur par défaut est JSON. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model. String : La clé du message est analysée directement comme une chaîne. JSON : La clé du message doit être au format JSON.	String
Message Value Format	Format d'entrée de la valeur du message. Les valeurs valides sont String et JSON. La valeur par défaut est JSON. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model. String : La valeur du message est analysée directement comme une chaîne. JSON : La valeur du message doit être au format JSON.	String
JSON Message Field Conversion	Méthode de traitement des champs d'un message JSON. Ce paramètre s'affiche si vous définissez Message Key Format ou Message Value Format sur JSON. Valeurs valides : Write All as String : Convertit tous les champs en type String dans Tablestore. Automatically Identify Field Types : Convertit les champs String et Boolean du corps du message JSON en types String et Boolean correspondants dans Tablestore. Les types de données Integer et Float du corps du message JSON sont convertis en type Double dans Tablestore.	Write All as String
Primary Key Mode	Spécifie le mode de clé primaire. Vous pouvez extraire les clés primaires de la table à partir de différentes parties des enregistrements de messages ApsaraMQ for Kafka, y compris les coordonnées (topic, partition et offset), la clé et la valeur. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model. La valeur par défaut est kafka. kafka : Utilise <connect_topic>_<connect_partition> et <connect_offset> comme clés primaires de la table de données. record_key : Utilise les champs de la clé d'enregistrement comme clés primaires de la table de données. record_value : Utilise les champs de la valeur d'enregistrement comme clés primaires de la table de données.	kafka
Primary Key Column Names	Noms des colonnes de clé primaire de la table de données et leurs types de données correspondants. Les types de données String et Integer sont pris en charge. Cela signifie que les champs de la clé ou de la valeur d'enregistrement correspondant aux noms de colonnes configurés sont utilisés comme clés primaires pour la table de données. Ce paramètre s'affiche si vous définissez Message Key Format sur JSON et Primary Key Mode sur record_key, ou si vous définissez Message Value Format sur JSON et Primary Key Mode sur record_value. Cliquez sur Create pour ajouter un nom de colonne. Vous pouvez configurer jusqu'à quatre noms de colonne.	None
Write Mode	Spécifie le mode d'écriture. Les valeurs valides sont put et update. La valeur par défaut est put. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model. put : Écrase les données existantes. update : Met à jour les données existantes.	put
Delete Mode	Si un enregistrement de message ApsaraMQ for Kafka contient des valeurs nulles, vous pouvez choisir de supprimer des lignes ou des colonnes d'attributs. Ce paramètre s'affiche lorsque Primary Key Mode est défini sur record_key. Valeurs valides : none : Valeur par défaut. Aucune suppression n'est autorisée. row : Autorise la suppression de lignes. column : Autorise la suppression de colonnes d'attributs. row_and_column : Autorise la suppression de lignes et de colonnes d'attributs. Le comportement de suppression dépend du mode d'écriture : Si le Write Mode est put, tout mode de suppression entraîne des écritures par écrasement dans la table de données Tablestore, même lorsque la valeur contient des champs nuls. Si le Write Mode est update et que le mode de suppression est none ou row, les enregistrements dont tous les champs de valeur sont nuls sont traités comme des données incorrectes. Si seuls certains champs de valeur sont nuls, le connecteur ignore les champs nuls et écrit les champs non nuls dans la table de données Tablestore. Si le mode de suppression est column ou row_and_column, le connecteur supprime les lignes et les colonnes d'attributs pour les champs nuls, puis écrit les données dans la table de données Tablestore.	None
Metric Name Field	Ce champ est mappé au champ de nom de mesure (_m_name) dans le TimeSeries Model de Tablestore. Le nom de mesure spécifie le nom de la grandeur physique ou de la métrique de surveillance pour les données de série temporelle, tel que la température ou la vitesse. Ce champ ne peut pas être vide. Ce paramètre s'affiche lorsque vous sélectionnez le TimeSeries Model pour Tablestore.	measurement
Data Source Field	Mappez ce champ au champ de source de données (_data_source) dans le TimeSeries Model de Tablestore. Ce champ sert d'identifiant pour la source des données de série temporelle, tel qu'un nom de machine ou un ID d'appareil, et peut être vide. Ce paramètre s'affiche lorsque le modèle TimeSeries est sélectionné pour Tablestore.	source
Tag Field	Utilisez un ou plusieurs champs comme champs de tag (_tags) pour un TimeSeries Model de Tablestore. Chaque tag est une paire clé-valeur de chaînes. La clé est le nom du champ configuré et la valeur est le contenu du champ. Les tags font partie des métadonnées de la série temporelle. Une série temporelle est identifiée de manière unique par la combinaison de son nom de métrique, de sa source de données et de ses tags. Les tags peuvent être vides. Ce paramètre s'affiche lorsque Tablestore est sélectionné comme modèle de série temporelle.	tag1, tag2
Timestamp Field	Mappe ce champ au champ d'horodatage (_time) dans le TimeSeries Model de Tablestore. Il représente le point dans le temps pour cette ligne de données de série temporelle, tel que le moment où une grandeur physique est générée. Lors de l'écriture des données dans Tablestore, le champ d'horodatage est converti en microsecondes pour l'écriture et le stockage. Ce paramètre s'affiche lorsque vous sélectionnez le TimeSeries Model pour Tablestore.	time
Timestamp Unit	Configurez ce paramètre en fonction du champ d'horodatage réel. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur TimeSeries Model. Valeurs valides : SECONDS MILLISECONDS MICROSECONDS NANOSECONDS	MILLISECONDS
Whether to Map All Non-primary Key Fields	Indique s'il faut mapper tous les champs hors clé primaire en tant que champs de données. Les champs hors clé primaire sont ceux qui ne sont pas déjà mappés comme nom de métrique, source de données, tag ou horodatage. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur TimeSeries Model. Valeurs valides : Yes : Les champs sont automatiquement mappés et leurs types de données sont déterminés. Les types numériques sont tous convertis en type Double. No : Vous devez spécifier les champs et les types à mapper.	Yes
Configure Mapping for All Non-primary Key Fields	Types de champs correspondant aux noms de champs hors clé primaire de la table de séries temporelles. Les types de données Double, Integer, String, Binary et Boolean sont pris en charge. Ce paramètre s'affiche si vous définissez Whether to Map All Non-primary Key Fields sur No.	String

Une fois le connecteur créé, vous pouvez le consulter sur la page Connectors.

Après la création du connecteur, recherchez-le sur la page Connectors et cliquez sur Deploy dans la colonne Actions.
Cliquez sur OK.

Envoyer un message de test

Après avoir déployé le connecteur sink Tablestore, envoyez un message au topic source dans ApsaraMQ for Kafka pour vérifier que les données sont bien synchronisées vers Tablestore.

Sur la page Connectors, localisez le connecteur cible et cliquez sur Test dans la colonne Actions.
Dans le panneau Send Message, envoyez un message de test.
- Définissez Sending Method sur Console.
  1. Dans la zone de texte Message Key, saisissez la clé du message. Par exemple, demo.
  2. Dans la zone de texte Message Content, saisissez le contenu du message de test. Par exemple, {"key": "test"}.
  3. Définissez Send to Specified Partition pour indiquer si le message doit être envoyé à une partition spécifique.
    - Cliquez sur Yes et saisissez l'ID de la partition dans la zone de texte Partition ID. Par exemple, 0. Pour obtenir l'ID de la partition, consultez Consulter l'état des partitions.
    - Cliquez sur No pour ne pas spécifier de partition.
- Définissez Sending Method sur Docker. Exécutez la commande Docker dans la section Run the Docker container to produce a sample message pour envoyer le message.
- Définissez Sending Method sur SDK. Sélectionnez un SDK pour le langage ou le framework requis ainsi qu'un type de connexion pour envoyer des messages.

Consulter les données de la table

Après avoir envoyé un message au topic source de données dans ApsaraMQ for Kafka, consultez les données dans la table Tablestore pour confirmer la réception du message.

Connectez-vous à la console Tablestore.
Sur la page Overview, cliquez sur le nom de l'instance ou cliquez sur Instances dans la colonne Actions.
Sous l'onglet Instance Details, localisez la table cible dans la section Tables.
Cliquez sur le nom de la table. Sur la page Table Manage, cliquez sur l'onglet Data Management pour consulter les données.