ApsaraMQ for Kafka:Créer un connecteur source MySQL

Fonctionnement

Lorsque vous créez un connecteur source MySQL via la console ApsaraMQ for Kafka, le système exécute automatiquement les actions suivantes :

1. Active DataWorks Basic Edition (gratuitement).

2. Crée un espace de travail DataWorks ainsi qu'un groupe de ressources exclusif pour l'intégration de données.

3. Génère les topics de destination dans votre instance Kafka, à raison d'un topic par table source.

Le groupe de ressources exclusif (4 vCPU, 8 Go de mémoire) exécute les tâches de synchronisation des données. Ces groupes de ressources fonctionnent sur abonnement mensuel et se renouvellent automatiquement à expiration.

<topic-prefix>_<source-table-name>

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• Une instance ApsaraMQ for Kafka avec la fonctionnalité connecteur activée, déployée dans l'une des régions suivantes : Chine (Shenzhen), Chine (Chengdu), Chine (Pékin), Chine (Zhangjiakou), Chine (Hangzhou), Chine (Shanghai) ou Singapour.

• Une instance ApsaraDB RDS for MySQL comprenant une base de données, un compte de base de données et au moins une table créée. Pour les références SQL, consultez Instructions SQL courantes pour MySQL.

• Des blocs CIDR ne se chevauchant pas entre le VPC de votre instance RDS MySQL et celui de votre instance Kafka.

• DataWorks autorisé à accéder à vos ENI via la page Cloud Resource Access Authorization.

• La source de données (instance ApsaraDB RDS for MySQL) et la destination de données (instance ApsaraMQ for Kafka) créées sous le même compte Alibaba Cloud.

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '<your-username>'@'%';

Créer et déployer le connecteur

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

2. Dans la section Resource Distribution de la page Overview, sélectionnez la région de votre instance.

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

4. Sur la page Connectors, sélectionnez votre instance dans la liste déroulante Select Instance, puis cliquez sur Create Connector.

5. Suivez l'assistant en trois étapes : cliquez sur Next. Step 2: Configure source service Sélectionnez ApsaraDB RDS for MySQL comme service source et configurez les paramètres suivants : cliquez sur Next. Step 3: Configure destination service Confirmez l'instance Kafka de destination et cliquez sur Create.

   Step 1: Configure basic information

   Paramètre	Description	Exemple
   Name	Nom unique du connecteur au sein de l'instance. De 1 à 48 caractères ; chiffres, lettres minuscules et traits d'union (-) autorisés. Ne peut pas commencer par un trait d'union. Le système crée automatiquement un groupe de consommateurs nommé connect-<connector-name>.	kafka-source-mysql
   Instance	Affiche le nom et l'ID de l'instance Kafka sélectionnée.	demo alikafka_post-cn-st21p8vj****

   Paramètre	Description	Exemple
   Region of ApsaraDB RDS for MySQL Instance	Région de l'instance RDS MySQL source.	China (Shenzhen)
   ApsaraDB RDS for MySQL Instance ID	ID de l'instance de la base de données source.	rm-wz91w3vk6owmz****
   Database Name	Nom de la base de données à synchroniser.	mysql-to-kafka
   Database Account	Nom d'utilisateur pour la connexion à la base de données.	mysql_to_kafka
   Password of Database Account	Mot de passe pour la connexion à la base de données.	--
   Database Table	Un ou plusieurs noms de tables, séparés par des virgules (,). Chaque table correspond à un topic.	mysql_tbl
   Tables to Automatically Add	Expression régulière permettant de détecter et synchroniser automatiquement les nouvelles tables. Utilisez .* pour inclure toutes les tables.	.*
   Topic Prefix	Préfixe pour les noms de topics générés automatiquement. Doit être globalement unique.	mysql

6. Sur la page Connectors, localisez le connecteur et cliquez sur Deploy dans la colonne Actions. Lorsque la colonne Status affiche Running, le connecteur est actif et synchronise les données.

Vérifier la synchronisation des données

1. Insérez une ligne de test dans une table source : pour plus d'exemples SQL, consultez Instructions SQL courantes pour MySQL.

      INSERT INTO mysql_tbl
          (mysql_title, mysql_author, submission_date)
          VALUES
          ("mysql2kafka", "tester", NOW());

2. Utilisez la fonctionnalité de requête de messages d'ApsaraMQ for Kafka pour confirmer l'arrivée des données dans le topic correspondant. Pour les instructions, consultez Requêter des messages. Un événement INSERT synchronisé avec succès se présente comme suit :

      {
          "schema": {
              "dataColumn": [
                  {
                      "name": "mysql_id",
                      "type": "LONG"
                  },
                  {
                      "name": "mysql_title",
                      "type": "STRING"
                  },
                  {
                      "name": "mysql_author",
                      "type": "STRING"
                  },
                  {
                      "name": "submission_date",
                      "type": "DATE"
                  }
              ],
              "primaryKey": [
                  "mysql_id"
              ],
              "source": {
                  "dbType": "MySQL",
                  "dbName": "mysql_to_kafka",
                  "tableName": "mysql_tbl"
              }
          },
          "payload": {
              "before": null,
              "after": {
                  "dataColumn": {
                      "mysql_title": "mysql2kafka",
                      "mysql_author": "tester",
                      "submission_date": 1614700800000
                  }
              },
              "sequenceId": "1614748790461000000",
              "timestamp": {
                  "eventTime": 1614748870000,
                  "systemTime": 1614748870925,
                  "checkpointTime": 1614748870000
              },
              "op": "INSERT",
              "ddl": null
          },
          "version": "0.0.1"
      }

Étapes suivantes

• Requêter des messages dans vos topics Kafka

• Activer la fonctionnalité connecteur pour d'autres instances