AnalyticDB:Synchroniser les données Kafka à l'aide de la fonctionnalité de synchronisation des données APS (Recommandé)

Limites

• Seuls les messages Kafka au format JSON sont pris en charge. Les autres formats provoquent des erreurs de synchronisation.

• Les modifications du schéma de table Kafka ne sont pas automatiquement propagées vers AnalyticDB for MySQL. Vous devez appliquer manuellement les modifications DDL.

• Les exemples de données Kafka supérieurs à 8 Ko sont tronqués par l'API Kafka, ce qui empêche le système d'analyser l'exemple et de générer automatiquement les mappages de champs.

• Les données écrites dans AnalyticDB for MySQL ne sont visibles qu'après l'exécution d'une opération de validation (Commit). L'intervalle de validation par défaut étant de 5 minutes, attendez au moins 5 minutes après le démarrage d'une tâche avant d'interroger le premier lot de données.

• Les chemins OSS utilisés dans différentes tâches de synchronisation ne peuvent pas partager le même préfixe. Par exemple, oss://testBucketName/test/sls1/ et oss://testBucketName/test/ entrent en conflit et entraînent l'écrasement des données.

• Si une tâche de synchronisation échoue et que les données du topic Kafka ont expiré, les données effacées ne peuvent pas être récupérées lors du redémarrage de la tâche. Pour réduire ce risque, augmentez la période de rétention des données du topic. En cas d'échec d'une tâche, contactez rapidement le support technique.

Prérequis

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

• Un cluster AnalyticDB for MySQL Enterprise Edition, Basic Edition ou Data Lakehouse Edition

• Un groupe de ressources pour les jobs

• Un compte de base de données pour le cluster (voir le tableau ci-dessous)

• Une instance ApsaraMQ for Kafka située dans la même région que le cluster AnalyticDB for MySQL

• Un topic Kafka contenant déjà des messages envoyés (voir Démarrage rapide pour ApsaraMQ for Kafka)

Exigences relatives aux comptes de base de données selon le type de compte :

Facturation

L'utilisation de la fonctionnalité de synchronisation des données entraîne les frais suivants :

• AnalyticDB for MySQL : Frais liés aux ressources ACU élastiques. Consultez Facturation pour Data Lakehouse Edition et Facturation pour Enterprise Edition et Basic Edition.

• Object Storage Service (OSS) : Frais de stockage, frais pour les requêtes GET, ainsi que frais pour les requêtes PUT et autres. Consultez Présentation de la facturation.

Fonctionnement

1. Ajoutez une source de données Kafka pour identifier l'instance et le topic ApsaraMQ for Kafka à lire.

2. Créez un lien de données (job de synchronisation) qui mappe les messages Kafka à une table AnalyticDB for MySQL, configure l'analyse JSON, les clés de partition et le décalage du consommateur.

3. Démarrez la tâche. Celle-ci commence à consommer les données Kafka à partir du décalage sélectionné et les écrit dans la table de destination.

4. Interrogez les données ingérées à l'aide de Spark SQL.

Étape 1 : Créer une source de données

Ignorez cette étape si vous avez déjà ajouté une source de données Kafka . Accédez directement à Étape 2 : Créer un lien de données .

1. Connectez-vous à la console AnalyticDB for MySQL. Dans le coin supérieur gauche, sélectionnez une région. Dans le volet de navigation de gauche, cliquez sur Clusters, puis cliquez sur l'ID du cluster.

2. Dans le volet de navigation de gauche, choisissez Data Ingestion > Data Sources.

3. Cliquez sur Create Data Source.

4. Sur la page Create Data Source, configurez les paramètres suivants :

   Paramètre	Description
   Data Source Type	Sélectionnez Kafka.
   Data Source Name	Généré automatiquement à partir du type de source et de l'heure actuelle. Modifiez-le si nécessaire.
   Data Source Description	Saisissez une description, telle que le scénario métier ou la portée des données.
   Deployment Mode	Seul le mode Alibaba Cloud Instance est pris en charge.
   Kafka Instance	L'ID de l'instance Kafka. Retrouvez-le sur la page Instances de la console ApsaraMQ for Kafka.
   Kafka Topic	Le nom du topic. Retrouvez-le sur la page Topics de l'instance de destination dans la console ApsaraMQ for Kafka.
   Message Data Format	Seul le format JSON est pris en charge.

5. Cliquez sur Create.

Étape 2 : Créer un lien de données

1. Dans le volet de navigation de gauche, cliquez sur Simple Log Service/Kafka Data Synchronization.

2. Cliquez sur Create Synchronization Job.

3. Sur la page Create Synchronization Job, configurez les trois sections ci-dessous.

Étape 3 : Démarrer la tâche de synchronisation des données

1. Sur la page Simple Log Service/Kafka Data Synchronization, localisez la tâche que vous avez créée et cliquez sur Start dans la colonne Actions.

2. Cliquez sur Search. La tâche a démarré avec succès lorsque son statut passe à Running.

Le premier lot de données est interrogeable au moins 5 minutes après le démarrage de la tâche, car les données sont validées par intervalles de 5 minutes par défaut.

Étape 4 : Analyser les données

Une fois les données synchronisées, utilisez Spark SQL pour les interroger dans AnalyticDB for MySQL. Pour plus d'informations, consultez Éditeur de développement Spark et Développement d'applications Spark hors ligne.

1. Dans le volet de navigation de gauche, choisissez Job Development > Spark JAR Development.

2. Saisissez vos instructions Spark SQL dans le modèle par défaut et cliquez sur Run Now. Voici un exemple :

   -- Example of Spark SQL. Modify the content and run your Spark program.
   
   conf spark.driver.resourceSpec=medium;
   conf spark.executor.instances=2;
   conf spark.executor.resourceSpec=medium;
   conf spark.app.name=Spark SQL Test;
   conf spark.adb.connectors=oss;
   
   -- SQL statements
   show tables from lakehouse20220413156_adbTest;

3. (Facultatif) Dans l'onglet Applications, cliquez sur Logs dans la colonne Actions pour afficher les journaux d'exécution Spark SQL.

Étape 5 (Facultative) : Gérer la source de données

Accédez à Data Ingestion > Data Sources. Les opérations suivantes sont disponibles dans la colonne Actions :

Niveaux d'analyse JSON et inférence de schéma

Le paramètre Parsed JSON Layers contrôle le nombre de niveaux imbriqués d'un message JSON développés en champs de destination distincts.

Exemple de message :

{
  "name" : "zhangle",
  "age" : 18,
  "device" : {
    "os" : {
        "test":lag,
        "member":{
             "fa":zhangsan,
             "mo":limei
           }
         },
    "brand" : "none",
    "version" : "11.4.2"
  }
}

Niveau 0 — Aucune analyse. L'intégralité du message JSON est sortie sous forme d'un seul champ.

Niveau 1 (par défaut) — Les champs de premier niveau sont développés.

Niveau 2 — Deux niveaux développés. Les champs non imbriqués sont sortis directement ; les champs imbriqués se développent en leurs sous-champs.

Niveau 3

Niveau 4

Étapes suivantes

• Éditeur de développement Spark

• Développement d'applications Spark hors ligne

• Facturation pour Data Lakehouse Edition

• Facturation pour Enterprise Edition et Basic Edition