Realtime Compute for Apache Flink:Ingestion de journaux en temps réel

Prérequis

• Assurez-vous que votre utilisateur RAM ou rôle RAM dispose des autorisations nécessaires pour accéder à la console Realtime Compute for Apache Flink. Pour plus d'informations, consultez la section Autorisations.

• Vous avez créé un espace de travail Flink. Pour plus d'informations, consultez la section Créer un espace de travail.

• Stockage amont et aval

  • Vous avez créé une instance ApsaraMQ for Kafka. Pour plus d'informations, consultez la section Étape 2 : Acheter et déployer une instance.

  • Vous avez créé une instance Hologres. Pour plus d'informations, consultez la section Acheter une instance Hologres.

  Remarque

  Vos instances ApsaraMQ for Kafka et Hologres doivent se trouver dans la même région et le même VPC que votre espace de travail Realtime Compute for Apache Flink. Dans le cas contraire, vous devez établir une connectivité réseau. Pour plus d'informations, consultez les sections Accéder aux services via des VPC ou Accéder à Internet.

Étape 1 : Configurer une liste blanche d'adresses IP

Pour permettre à Flink d'accéder à vos instances Kafka et Hologres, ajoutez le bloc CIDR de votre espace de travail Flink aux listes blanches d'adresses IP de Kafka et Hologres.

1. Obtenez le bloc CIDR du VPC de votre espace de travail Flink.

   1. Connectez-vous à la console Realtime Compute for Apache Flink.

   2. Dans la colonne Actions de l'workspace cible, choisissez More > Workspace Details.

   3. Dans la boîte de dialogue Workspace Details, recherchez le CIDR block du vSwitch.

      La boîte de dialogue affiche les informations de base de l'espace de travail ainsi qu'une liste de vSwitches. Dans la colonne Bloc CIDR, consultez le bloc CIDR correspondant à chaque zone de disponibilité. Notez ce bloc CIDR pour l'étape suivante.

2. Ajoutez le bloc CIDR de l'espace de travail Flink à la liste blanche d'adresses IP de votre instance Kafka.

   Vous devez configurer la liste blanche d'adresses IP pour un endpoint VPC. Pour connaître la procédure détaillée, consultez la section Configurer une liste blanche d'adresses IP. Dans la boîte de dialogue de modification de la liste blanche, cliquez sur Ajouter une adresse IP à la liste blanche pour ajouter le bloc CIDR.

3. Ajoutez le bloc CIDR de l'espace de travail Flink à la liste blanche d'adresses IP de votre instance Hologres.

   Connectez-vous à votre instance Hologres et configurez sa liste blanche d'adresses IP. Pour connaître la procédure détaillée, consultez la section Liste blanche d'adresses IP. Sur la page de configuration de la liste blanche dans le Security Center de HoloWeb, saisissez le bloc CIDR dans le champ Adresse IP de la boîte de dialogue Modifier la liste blanche d'adresses IP, puis cliquez sur OK.

Étape 2 : Préparer les données de test Kafka

Utilisez le connecteur Faker de Realtime Compute for Apache Flink pour générer des données et les écrire dans ApsaraMQ for Kafka. Suivez les étapes ci-dessous dans la console de développement de Realtime Compute for Apache Flink.

1. Dans la console ApsaraMQ for Kafka, créez un topic nommé users.

   Pour plus d'informations, consultez la section Créer un topic.

2. Créez une tâche pour écrire des données dans ApsaraMQ for Kafka.

   1. Connectez-vous à la console de développement de Realtime Compute for Apache Flink.

   2. Dans la colonne Actions de l'espace de travail cible, cliquez sur Console.

   3. Dans le volet de navigation de gauche, choisissez Development > ETL.

   4. Cliquez sur l'icône , puis cliquez sur New Blank Stream Draft. Saisissez un Name et sélectionnez une Engine Version.

      Realtime Compute for Apache Flink propose également divers modèles de code et de synchronisation de données. Chacun inclut des cas d'utilisation spécifiques, des exemples de code et des instructions. Cliquez sur un modèle pour apprendre rapidement les fonctionnalités et la syntaxe de Realtime Compute for Apache Flink et implémenter votre logique métier. Pour plus d'informations, consultez les sections Modèles de code et Modèles de synchronisation de données.

      Paramètre	Description	Exemple
      Name	Le nom de la tâche. Remarque Le nom de la tâche doit être unique au sein de l'espace de travail actuel.	flink-test
      Engine Version	La version du moteur Flink pour la tâche actuelle. Sélectionnez une version étiquetée Recommended ou Stable pour une fiabilité et des performances accrues. Pour plus d'informations sur les versions du moteur, consultez les sections Notes de version et Versions du moteur.	vvr-8.0.8-flink-1.17

   5. Cliquez sur Create.

   6. Rédigez les instructions SQL pour la tâche.

      Copiez le code suivant dans l'éditeur et modifiez les paramètres en fonction de votre environnement.

      CREATE TEMPORARY TABLE source (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        event_time TIMESTAMP
      ) WITH (
        'connector' = 'faker',
        'number-of-rows' = '100',
        'rows-per-second' = '10',
        'fields.id.expression' = '#{number.numberBetween ''0'',''1000''}',
        'fields.first_name.expression' = '#{name.firstName}',
        'fields.last_name.expression' = '#{name.lastName}',
        'fields.address.country.expression' = '#{Address.country}',
        'fields.address.state.expression' = '#{Address.state}',
        'fields.address.city.expression' = '#{Address.city}',
        'fields.event_time.expression' = '#{date.past ''15'',''SECONDS''}'
      );
      
      CREATE TEMPORARY TABLE sink (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        `timestamp` TIMESTAMP METADATA
      ) WITH (
        'connector' = 'kafka',
        'properties.bootstrap.servers' = 'alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092',
        'topic' = 'users',
        'format' = 'json',
        'properties.enable.idempotence'='false'
      );
      
      INSERT INTO sink SELECT * FROM source;

      Le tableau suivant décrit les paramètres que vous devez modifier.

      Paramètre	Valeur d'exemple	Description
      properties.bootstrap.servers	alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092	Les endpoints des brokers ApsaraMQ for Kafka. Une liste d'endpoints de brokers séparés par des virgules au format « hôte:port ». Vous pouvez trouver l'endpoint Domain Name du VPC dans la section Endpoint Information de la page Instance Details.
      topic	users	Le nom du topic ApsaraMQ for Kafka.

3. Démarrez la tâche.

   1. Sur la page Development > ETL, cliquez sur Deploy.

   2. Dans la boîte de dialogue Deploy draft, cliquez sur Confirm.

   3. Configurez les ressources pour la tâche. Pour plus d'informations, consultez la section Configurer les ressources d'une tâche.

   4. Sur la page O&M > Deployments, recherchez le déploiement cible et cliquez sur Start dans la colonne Actions. Pour plus d'informations sur la configuration de démarrage, consultez la section Démarrer un déploiement.

   5. Sur la page Deployments, vous pouvez surveiller les informations d'exécution et l'état du déploiement.

      Étant donné que la source Faker génère un flux borné, l'état du déploiement passe à FINISHED environ une minute après le démarrage. Une fois le déploiement terminé, les données ont été écrites dans le topic « users ». Le code suivant présente un exemple de données au format JSON.

      {
        "id": 765,
        "first_name": "Barry",
        "last_name": "Pollich",
        "address": {
          "country": "United Arab Emirates",
          "state": "Nevada",
          "city": "Powlowskifurt"
        }
      }

Étape 3 : Créer et démarrer une tâche de synchronisation de données

Étape 4 : Afficher le résultat complet de la synchronisation

1. Connectez-vous à la console de gestion Hologres.

2. Sur la page Network Information, cliquez sur le nom de votre instance cible.

3. Dans le coin supérieur droit de la page, cliquez sur instance details.

4. Sur l'onglet Save, affichez le schéma de table et les données de la table users synchronisée dans la base de données flink_test_db.

   Dans l'arborescence de navigation de gauche, développez le nom de votre instance > flink_test_db > test_schema > Tables. La table users synchronisée apparaît.

   Le schéma et les données de la table synchronisée sont les suivants.

   • Schéma de table

     Double-cliquez sur le nom de la table users pour afficher le schéma de table.

     Le schéma de la table users contient les champs suivants : id (BIGINT, clé primaire), first_name (TEXT), last_name (TEXT), address.country (TEXT), address.state (TEXT) et address.city (TEXT).

     Remarque

     Lors de la synchronisation complète des données, nous vous recommandons de définir la partition et le décalage (offset) des métadonnées Kafka comme clé primaire pour la table Hologres. Cela évite la duplication des données en cas de basculement de la tâche et de retransmission des données.

   • Données de table

     Dans le coin supérieur droit de la page de la table users, cliquez sur Development. Saisissez la commande suivante et cliquez sur ETL.

     SELECT * FROM test_schema.users;

     La commande renvoie le résultat suivant.

     La requête renvoie plusieurs lignes. Cela confirme que les enregistrements ont été synchronisés avec succès vers la table users. Chaque ligne renvoyée contient les données complètes pour les colonnes id, first_name, last_name, address.country, address.state et address.city.

Étape 5 : Observer la synchronisation automatique du schéma

1. Dans la console ApsaraMQ for Kafka, envoyez manuellement un message contenant une nouvelle colonne.

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

   2. Sur la page Deploy, cliquez sur le nom de l'instance cible.

   3. Sur la page O&M, cliquez sur le nom du topic cible (users).

   4. Cliquez sur Deployments.

   5. Configurez le message.

      Dans la boîte de dialogue Commencer à envoyer et consommer un message, configurez les paramètres comme suit.

      Paramètre	Exemple
      Start	Sélectionnez Actions.
      Instances	Saisissez flinktest.
      Connect to Instance	Copiez et collez le contenu JSON suivant dans le champ Contenu du message. { "id": 100001, "first_name": "Dennise", "last_name": "Schuppe", "address": { "country": "Isle of Man", "state": "Montana", "city": "East Coleburgh" }, "house-points": { "house": "Pukwudgie", "points": 76 } } Remarque Dans cet exemple, house-points est une nouvelle colonne imbriquée.
      Metadata Management	Sélectionnez Query table.
      Run	Saisissez 0.

   6. Cliquez sur Instances.

2. Dans la console Hologres, affichez les modifications de schéma et de données dans la table users.

   1. Connectez-vous à la console Hologres.

   2. Sur la page Topics, cliquez sur le nom de l'instance cible.

   3. Dans le coin supérieur droit de la page, cliquez sur Send Message.

   4. Sur l'onglet Method of Sending, double-cliquez sur le nom de la table users.

   5. Cliquez sur Console, saisissez l'instruction suivante, puis cliquez sur Message Key.

      SELECT * FROM test_schema.users;

   6. Affichez les résultats de la requête.

      La requête renvoie le résultat suivant.

      Le résultat montre que l'enregistrement dont l'ID est 100001 a bien été écrit dans Hologres, et que deux nouvelles colonnes, house-points.house et house-points.points, ont été ajoutées à la table Hologres.

      Remarque

      Le message envoyé à ApsaraMQ for Kafka ne contient qu'une seule colonne imbriquée : house-points. Toutefois, étant donné que json.infer-schema.flatten-nested-columns.enable est spécifié dans la clause WITH, Realtime Compute for Apache Flink aplatit automatiquement cette colonne, en utilisant les chemins d'accès aux champs imbriqués comme noms de nouvelles colonnes.

Références

• Pour savoir comment utiliser Message Queue for Apache Kafka comme table source ou table de résultats, consultez la section Message Queue for Apache Kafka.

• Pour améliorer les performances des tâches en ajustant le parallélisme des nœuds et les ressources, consultez la section Configurer le déploiement d'une tâche.