ApsaraMQ for Kafka:Créer un connecteur de réception AnalyticDB

Prérequis

Avant de commencer, assurez-vous que les prérequis suivants sont remplis :

• Message Queue for Apache Kafka

  • Activez la fonctionnalité de connecteur pour l'instance Message Queue for Apache Kafka. Pour plus d'informations, consultez la section Activer les connecteurs.

  • Créez un sujet source pour l'instance Message Queue for Apache Kafka. Pour plus d'informations, consultez la section Étape 1 : Créer un sujet.

• Function Compute

  • Activer Function Compute.

• AnalyticDB for MySQL et AnalyticDB for PostgreSQL

  • AnalyticDB for MySQL : Dans la console AnalyticDB for MySQL, créez un cluster, créez un compte de base de données, connectez-vous au cluster, puis créez une base de données. Pour plus d'informations, consultez les sections Créer un cluster, Créer un compte de base de données, Se connecter à un cluster et Créer une base de données.

  • AnalyticDB for PostgreSQL : Dans la console AnalyticDB for PostgreSQL, créez une instance, créez un compte de base de données, puis connectez-vous à la base de données. Pour plus d'informations, consultez les sections Créer une instance, Créer et gérer des comptes et Se connecter à une instance.

Notes d'utilisation

• Vous ne pouvez exporter les données d'un sujet source d'une instance Message Queue for Apache Kafka vers AnalyticDB for MySQL ou AnalyticDB for PostgreSQL via AnalyticDB for MySQL que si tous les services se trouvent dans la même région. Pour plus d'informations sur les limites des connecteurs, consultez la section Limites.

• Cette fonctionnalité est alimentée par Function Compute. Function Compute propose un quota gratuit. Des frais s'appliquent pour l'utilisation des ressources dépassant ce quota. Pour plus d'informations, consultez la section Présentation de la facturation de Function Compute.

• Vous pouvez interroger les journaux d'appels de fonction dans Function Compute pour résoudre les problèmes. Pour plus d'informations, consultez la section Configurer la journalisation.

• Lors de l'exportation des données, Message Queue for Apache Kafka sérialise les messages sous forme de chaînes UTF-8. Les formats de données binaires ne sont pas pris en charge.

• Si le connecteur de réception AnalyticDB utilise un endpoint privé, l'environnement d'exécution Function Compute ne peut pas y accéder par défaut. Pour activer la connectivité réseau, configurez le service de fonction dans la console Function Compute avec le même VPC et le même vSwitch que votre instance AnalyticDB de destination. Pour plus d'informations, consultez la section Mettre à jour un service.

• Message Queue for Apache Kafka crée automatiquement un rôle lié au service lorsque vous créez un connecteur.

  • Si aucun rôle lié au service n'existe, Message Queue for Apache Kafka en crée un automatiquement. Ce rôle accorde à Message Queue for Apache Kafka les autorisations nécessaires pour exporter les données.

  • Si un rôle lié au service existe déjà, Message Queue for Apache Kafka n'en crée pas un autre.

  Pour plus d'informations sur les rôles liés au service, consultez la section Rôles liés au service.

Procédure

Pour exporter les données d'un sujet source d'une instance Message Queue for Apache Kafka vers une base de données AnalyticDB à l'aide d'un connecteur de réception AnalyticDB, suivez les étapes ci-dessous :

1. Facultatif : Créez le sujet et le group requis pour le connecteur de réception AnalyticDB

   Si vous n'avez pas besoin de personnaliser le sujet et le Group, vous pouvez ignorer cette étape et choisir de les créer automatiquement à l'étape suivante.

   Important

   Certains sujets requis par le connecteur de réception AnalyticDB doivent utiliser le stockage local. Pour les instances Message Queue for Apache Kafka exécutant la version majeure 0.10.2, vous ne pouvez pas créer manuellement des sujets utilisant le stockage local. Ces sujets ne peuvent être créés qu'automatiquement.

   1. Créer les sujets requis par un connecteur de réception AnalyticDB

   2. Créer le group requis par un connecteur de réception AnalyticDB

2. Créer et déployer un connecteur de réception AnalyticDB

3. Configurer les services

   1. Configurer le service Function Compute

   2. Configurer la base de données AnalyticDB de destination

4. Vérifier le résultat

   1. Envoyer des messages de test

   2. Vérifier le résultat

Créer les sujets requis

Vous pouvez créer manuellement les cinq sujets requis par le connecteur de réception AnalyticDB dans la console Message Queue for Apache Kafka : un sujet d'offset de tâche, un sujet de configuration de tâche, un sujet d'état de tâche, un sujet de file d'attente de lettres mortes et un sujet de données d'erreur. Ces sujets ont des exigences différentes concernant le nombre de partitions et le moteur de stockage. Pour plus d'informations, consultez le tableau Paramètres de configuration du service source.

1. Connectez-vous à la console ApsaraMQ for Kafka.

2. Sur la page Overview, sélectionnez une région dans la section Resource Distribution.

   Important

   Vous devez créer les sujets dans la même région que votre application, c'est-à-dire la région où l'instance ECS est déployée. Les sujets ne peuvent pas être utilisés entre différentes régions. Par exemple, si un sujet 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 dans la région Chine (Pékin).

3. Sur la page Instances, cliquez sur le nom de l'instance cible.

4. Dans le volet de navigation de gauche, cliquez sur Topics.

5. Sur la page Topics, cliquez sur Create Topic.

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

   Paramètre	Description	Exemple
   Name	Le nom du sujet. Remarque Dans Kafka, les noms de sujets tels que xxx_xxx et xxx.xxx sont considérés comme identiques. Une erreur est signalée si vous tentez de créer un sujet avec un nom dupliqué.	demo
   Description	Une brève description du sujet.	test demo
   Partitions	Le nombre de partitions dans le sujet.	12
   Storage Engine Remarque Actuellement, vous pouvez sélectionner un type de moteur de stockage uniquement pour les instances Professional Edition non Serverless. Pour les autres instances, ce paramètre n'est pas pris en charge et est défini par défaut sur Cloud Storage.	Le moteur de stockage pour les messages du sujet. 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 durabilité élevée et une fiabilité élevée. Il utilise un mécanisme distribué à trois réplicas. Si l'Instance Edition de l'instance est Standard (High Write), le moteur de stockage ne peut être que Cloud Storage. Local Storage : Utilise l'algorithme de réplication ISR (In-Sync Replica) natif de Kafka et un mécanisme distribué à trois réplicas.	Cloud Storage
   Message Type	Le type de messages du sujet. Normal Message : Par défaut, les messages ayant la même clé sont distribués vers la même partition, et les messages au sein d'une partition sont stockés dans l'ordre de leur envoi. Si une machine du cluster tombe en panne, les messages peuvent devenir désordonnés. Lorsque vous définissez Storage Engine sur Cloud Storage, Normal Message est sélectionné par défaut. Partitionally Ordered Message : Par défaut, les messages ayant la même clé sont distribués vers la même partition, et les messages au sein d'une partition sont stockés dans l'ordre de leur envoi. Si une machine du cluster tombe en panne, l'ordre des messages est toujours garanti au sein de la partition. Cependant, l'envoi de messages vers certaines partitions peut échouer. Les opérations normales reprennent après la récupération des partitions. Lorsque vous définissez Storage Engine sur Local Storage, Partitionally Ordered Message est sélectionné par défaut.	Normal Message
   Log Cleanup Policy	La politique de nettoyage des journaux du sujet. Lorsque vous définissez Storage Engine sur Local Storage, vous devez configurer la Log Cleanup Policy. Vous ne pouvez définir le moteur de stockage sur Local Storage 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 pour garantir la disponibilité du service. Compact : Utilise la politique de nettoyage Kafka Log Compaction. La politique Log Compaction garantit que pour les messages ayant la même clé, la dernière valeur est toujours conservée. Cela s'applique principalement aux scénarios tels que la récupération de l'é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 Confluent Schema Registry, vous devez utiliser un sujet Kafka Compact pour stocker l'état du système ou les informations de configuration. Important Les sujets Compact 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 sujets dans d'autres scénarios d'envoi et de réception de messages. Pour plus d'informations, consultez la bibliothèque de démonstration ApsaraMQ for Kafka.	Compact
   Tag	Les tags du sujet.	demo

   Après la création du sujet, il s'affiche dans la liste sur la page Topics.

Créer le group requis par le connecteur de réception AnalyticDB

Vous pouvez créer manuellement le group pour la tâche de synchronisation des données du connecteur de réception AnalyticDB dans la console Message Queue for Apache Kafka. Le nom du group doit être connect-Nom de la tâche. Pour plus d'informations, consultez la section Paramètres du service source.

1. Connectez-vous à la console ApsaraMQ for Kafka.

2. Sur la page Overview, sélectionnez une région dans la section Resource Distribution.

3. Sur la page Instances, cliquez sur le nom de l'instance cible.

4. Dans le volet de navigation de gauche, cliquez sur Groups.

5. Sur la page Groups, cliquez sur Create Group.

6. 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.

   Après la création du group, il s'affiche dans la liste sur la page Groups.

Créer et déployer un connecteur de sortie AnalyticDB

1. Connectez-vous à la console ApsaraMQ for Kafka.

2. Sur la page Overview, sélectionnez une région dans la section Resource Distribution.

3. Dans le volet de navigation de gauche, cliquez sur Connectors.

4. Sur la page Connectors, sélectionnez l'instance à laquelle appartient le connecteur dans la liste déroulante Select Instance et cliquez sur Create Connector.

5. Dans l'assistant Create Connector, effectuez les étapes suivantes.

   1. Sous l'onglet Configure Basic Information, configurez les paramètres selon vos besoins, puis cliquez sur Next.

      Parameter	Description	Example value
      Name	Nom du connecteur. Règles de nommage : Le nom doit comporter entre 1 et 48 caractères. Le premier caractère peut être une lettre majuscule, une lettre minuscule ou un chiffre. Les caractères suivants peuvent être des lettres minuscules, des chiffres ou des traits d'union (-). Le nom doit être unique au sein d'une même instance Message Queue for Apache Kafka. Une tâche de synchronisation des données pour un Connecteur doit utiliser un Group nommé connect-nom de la tâche. Si vous ne créez pas manuellement le Group, le système le créera automatiquement pour vous.	kafka-adb-sink
      Instance	Le nom et l'ID de l'instance sont renseignés automatiquement.	demo alikafka_post-cn-st21p8vj****

   2. Sous l'onglet Configure Source Service, sélectionnez Message Queue for Apache Kafka comme Data Source, configurez les paramètres suivants, puis cliquez sur Next.

      Tableau 1. Paramètres du service source

      Parameter	Description	Example
      Data Source Topic	Rubrique à partir de laquelle synchroniser les données.	adb-test-input
      Consumer Thread Concurrency	Nombre de threads de consommateur simultanés pour la rubrique source. Valeur par défaut : 6. Valeurs valides : 1 2 3 6 12	6
      Consumer Offset	Position à partir de laquelle la consommation commence. Valeurs valides : Earliest Offset : démarre la consommation à partir du décalage disponible le plus ancien. Latest Offset : démarre la consommation à partir du décalage disponible le plus récent.	Earliest Offset
      VPC ID	VPC où se trouve la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Par défaut, il s'agit du VPC de l'instance Message Queue for Apache Kafka et ne nécessite aucune saisie manuelle.	vpc-bp1xpdnd3l***
      vSwitch ID	vSwitch où se trouve la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Ce vSwitch doit se trouver dans le même VPC que l'instance Message Queue for Apache Kafka. Par défaut, il s'agit du vSwitch spécifié lors du déploiement de l'instance Message Queue for Apache Kafka.	vsw-bp1d2jgg81***
      Failure Handling Policy	Politique de gestion des partitions en cas d'échec d'envoi d'un message. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Valeurs valides : Continue Subscription : poursuit l'abonnement à la partition et imprime un journal d'erreurs. Stop Subscription : arrête l'abonnement à la partition et imprime un journal d'erreurs. Remarque Pour savoir comment afficher les journaux, consultez Opérations liées aux connecteurs. Pour trouver des solutions basées sur les codes d'erreur, consultez Codes d'erreur.	Continue Subscription
      Resource Creation Method	Méthode de création des rubriques et du groupe requis par le connecteur. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Auto Manual	Auto
      Connector Consumer Group	Groupe utilisé par la tâche de synchronisation des données du connecteur. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Le nom du groupe doit être connect-nom de la tâche.	connect-kafka-adb-sink
      Task Offset Topic	Rubrique utilisée pour stocker les décalages des consommateurs. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Rubrique : il est recommandé que le nom commence par connect-offset. Partitions : le nombre de partitions doit être supérieur à 1. Moteur de stockage : la rubrique doit utiliser local storage. cleanup.policy : la politique de nettoyage des journaux de la rubrique doit être compact.	connect-offset-kafka-adb-sink
      Task Configuration Topic	Rubrique utilisée pour stocker les configurations des tâches. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Rubrique : il est recommandé que le nom commence par connect-config. Partitions : le nombre de partitions doit être égal à 1. Moteur de stockage : la rubrique doit utiliser local storage. cleanup.policy : la politique de nettoyage des journaux de la rubrique doit être compact.	connect-config-kafka-adb-sink
      Task Status Topic	Rubrique utilisée pour stocker l'état des tâches. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Rubrique : il est recommandé que le nom commence par connect-status. Partitions : le nombre recommandé de partitions est 6. Moteur de stockage : la rubrique doit utiliser local storage. cleanup.policy : la politique de nettoyage des journaux de la rubrique doit être compact.	connect-status-kafka-adb-sink
      Dead-letter Queue Topic	Rubrique utilisée pour stocker les données d'erreur provenant du framework Connect. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Cette rubrique peut être identique à la rubrique de données d'erreur afin d'économiser des ressources de rubrique. Rubrique : il est recommandé que le nom commence par connect-error. Partitions : le nombre recommandé de partitions est 6. Moteur de stockage : la rubrique peut utiliser local storage ou cloud storage.	connect-error-kafka-adb-sink
      Error Data Topic	Rubrique utilisée pour stocker les données d'exception provenant du Sink. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Cette rubrique peut être identique à la rubrique de file d'attente de lettres mortes afin d'économiser des ressources de rubrique. Rubrique : il est recommandé que le nom commence par connect-error. Partitions : le nombre recommandé de partitions est 6. Moteur de stockage : la rubrique peut utiliser local storage ou cloud storage.	connect-error-kafka-adb-sink

   3. Sous l'onglet Configure Destination Service, sélectionnez AnalyticDB comme Destination Service, configurez les paramètres suivants, puis cliquez sur Create.

      Parameter	Description	Example
      Instance Type	Type de l'instance AnalyticDB de destination. Valeurs valides : AnalyticDB for MySQL et AnalyticDB for PostgreSQL.	MySQL
      AnalyticDB Instance ID	ID de l'instance de destination AnalyticDB for MySQL ou AnalyticDB for PostgreSQL.	am-bp139yqk8u1ik****
      Database Name	Nom de la base de données de destination dans l'instance AnalyticDB.	adb_demo
      Table Name	Nom de la table de destination dans la base de données AnalyticDB pour le stockage des messages.	user
      Database Username	Nom d'utilisateur pour la connexion à l'instance AnalyticDB afin d'importer les données.	adbmysql
      Database Password	Mot de passe pour la connexion à l'instance AnalyticDB afin d'importer les données. Vous définissez le mot de passe lors de la création de l'instance. Si vous oubliez le mot de passe, vous pouvez le réinitialiser. AnalyticDB for MySQL : pour plus d'informations sur la réinitialisation du mot de passe, consultez Modifier le mot de passe d'un compte. AnalyticDB for PostgreSQL : connectez-vous à la console AnalyticDB for PostgreSQL, cliquez sur le nom de l'instance, puis cliquez sur Account Management dans le volet de navigation de gauche. Recherchez le compte dont vous souhaitez réinitialiser le mot de passe et cliquez sur Reset Password dans la colonne Actions.	****

      Remarque

      Message Queue for Apache Kafka transmet le nom d'utilisateur et le mot de passe en tant que variables d'environnement à la fonction Function Compute lors de la création de la tâche. Une fois la tâche créée, Message Queue for Apache Kafka ne conserve pas ces informations.

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

6. Sur la page Connectors, recherchez le connecteur que vous avez créé et cliquez sur Deploy dans la colonne Actions.

Configurer le service Function Compute

Après avoir créé et déployé avec succès un connecteur de sortie AnalyticDB dans la console Message Queue for Apache Kafka, Function Compute crée automatiquement un service et une fonction pour le connecteur. Le service est nommé selon le format kafka-service-<nom_du_connecteur>-<chaîne_aléatoire>, et la fonction est nommée selon le format fc-adb-<chaîne_aléatoire>.

1. Sur la page Connectors, recherchez le connecteur cible et cliquez sur Configure Function dans la colonne Actions.

   Cette action vous redirige vers la console Function Compute.

2. Dans la console Function Compute, recherchez le service de fonction créé automatiquement et configurez ses informations VPC et vSwitch. Pour connaître la procédure détaillée, consultez Mettre à jour un service.

Configurer la base de données AnalyticDB de destination

Après avoir configuré le service Function Compute, vous devez ajouter le bloc CIDR de son vSwitch à la liste d'autorisation de votre instance AnalyticDB. Vous trouverez ce bloc CIDR sur la page vSwitch de la console VPC, dans la ligne correspondant au VPC et au vSwitch du service Function Compute.

• AnalyticDB for MySQL : connectez-vous à la console AnalyticDB for MySQL et configurez la liste d'autorisation. Pour plus d'informations, consultez Configurer une liste d'autorisation d'adresses IP.

• AnalyticDB for PostgreSQL : connectez-vous à la console AnalyticDB for PostgreSQL et configurez la liste d'autorisation. Pour plus d'informations, consultez Configurer une liste d'autorisation.

Envoyer des messages de test

1. Sur la page Connectors, recherchez le connecteur cible et cliquez sur Test dans la colonne Actions.

2. 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 partition dans la zone de texte Partition ID. Par exemple, 0. Pour interroger l'ID de partition, consultez Afficher l'état de la partition.

        • 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 et un type de connexion pour envoyer les messages.

Vérifier le résultat

Après avoir envoyé des messages à la rubrique source dans Message Queue for Apache Kafka, connectez-vous à la console AnalyticDB for MySQL ou à la console AnalyticDB for PostgreSQL, connectez-vous à la base de données et accédez à la fenêtre SQL Window de Data Management Service (DMS). Interrogez la table dans l'instance correspondante pour vérifier si les données ont été correctement exportées.