Realtime Compute for Apache Flink:Job Flink SQL

Prérequis

• L'utilisateur RAM ou le rôle RAM que vous utilisez dispose des autorisations requises pour la console Realtime Compute. Pour plus d'informations, consultez Autorisations.

• Un espace de travail Flink a été créé. Pour plus d'informations, consultez Activer Realtime Compute for Apache Flink.

Étape 1 : Créer un brouillon SQL

1. Accédez à la page de création du brouillon SQL.

   1. Connectez-vous à la console Realtime Compute.

   2. Recherchez l'espace de travail Flink cible et cliquez sur Console dans la colonne Actions.

   3. Dans le volet de navigation, cliquez sur Development > ETL.

2. 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 divers modèles de code et de synchronisation des données. Chaque modèle comprend des descriptions de cas d'utilisation, des exemples de code et des instructions. Cliquez sur un modèle pour découvrir rapidement les fonctionnalités du produit et la syntaxe afin de mettre en œuvre votre logique métier. Pour plus d'informations, consultez Modèles de code et Modèles de synchronisation des données.

   Paramètre	Description	Exemple
   Name	Nom du brouillon SQL. Remarque Le nom doit être unique au sein du projet actuel.	flink-test
   Engine Version	Version du moteur Flink pour le brouillon SQL. Nous vous recommandons d'utiliser les versions portant l'étiquette RECOMMENDED ou Stable pour une meilleure fiabilité et de meilleures performances. Pour plus d'informations sur les versions du moteur, consultez les Notes de version et les Versions du moteur.	vvr-8.0.8-flink-1.17

3. Cliquez sur Create.

Étape 2 : Rédiger le code SQL et afficher les configurations du brouillon

1. Rédigez le code SQL.

   Copiez le code SQL suivant dans l'éditeur. Cet exemple utilise un connecteur Datagen pour générer un flux de données aléatoire et un connecteur Print pour écrire la sortie dans les journaux de la console. Pour plus d'informations sur les connecteurs pris en charge, consultez Connecteurs pris en charge.

   -- Create a temporary source table named datagen_source.
   CREATE TEMPORARY TABLE datagen_source(
     randstr VARCHAR
   ) WITH (
     'connector' = 'datagen' -- Use the Datagen connector.
   );
   
   -- Create a temporary sink table named print_table.
   CREATE TEMPORARY TABLE print_table(
     randstr  VARCHAR
   ) WITH (
     'connector' = 'print',   -- Use the Print connector.
     'logger' = 'true'        -- Write the output to logs.
   );
   
   -- Select a substring from the randstr field and insert it into the sink table.
   INSERT INTO print_table
   SELECT SUBSTRING(randstr,0,8) from datagen_source;

   Remarque

   • Cet exemple utilise une instruction INSERT INTO pour écrire des données dans une seule table de destination (sink). Vous pouvez également utiliser l'instruction INSERT INTO pour écrire des données dans plusieurs tables de destination. Pour plus d'informations, consultez Instruction INSERT INTO.

   • En production, utilisez les tables enregistrées dans les Catalogs au lieu des tables temporaires. Pour plus d'informations, consultez Catalogs.

2. Consultez les configurations du brouillon.

   À droite de l'éditeur SQL, vous pouvez afficher ou configurer les paramètres dans plusieurs onglets.

   Onglet	Description
   Configurations	Engine version : pour plus d'informations, consultez les Versions du moteur et les Politiques de cycle de vie. Nous vous recommandons d'utiliser une version recommandée ou stable. Les étiquettes de version sont les suivantes : Recommend : dernière version mineure de la version majeure actuelle. Stable : dernière version mineure d'une version majeure au cours de sa période de service, avec les défauts connus corrigés. Normal : autres versions mineures toujours dans leur période de service. Deprecated : versions ayant dépassé leur date de fin de vie (EOL). Additional dependencies : dépendances supplémentaires requises pour le job, telles que les fonctions temporaires. Authentification Kerberos : activez l'authentification Kerberos et configurez un cluster Kerberos enregistré ainsi que les informations de principal. Si vous n'avez pas enregistré de cluster Kerberos, consultez Enregistrer un cluster Hive Kerberos.
   Structure	Flow Diagram : utilisez le diagramme de flux de données pour afficher rapidement la lignée des données. Tree Diagram : utilisez le diagramme arborescent pour afficher rapidement les sources de données.
   Versions	Vous pouvez consulter ici l'historique des versions du brouillon SQL. Pour plus d'informations sur les fonctions de la colonne Actions, consultez Gérer les versions des brouillons.

(Facultatif) Étape 3 : Valider et déboguer le brouillon SQL

1. Validez le brouillon SQL.

   La validation vérifie la sémantique SQL, la connectivité réseau et les métadonnées des tables utilisées dans votre brouillon. Après la validation, cliquez sur SQL Advice dans la zone de résultats pour afficher les risques SQL potentiels et les suggestions d'optimisation.

   1. Dans le coin supérieur droit de l'éditeur SQL, cliquez sur Validate.

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

   Remarque

   Si une erreur de délai d'expiration se produit, vous pourriez voir le message suivant : The RPC times out maybe because the SQL parsing is too complicated. Please consider enlarging the flink.sqlserver.rpc.execution.timeout option in flink-configuration, which by default is 120 s.

   Solution : ajoutez le paramètre de configuration suivant en haut de votre éditeur SQL.

   SET 'flink.sqlserver.rpc.execution.timeout' = '600s';

2. Déboguez le brouillon SQL.

   La fonctionnalité de débogage vous permet de simuler l'exécution d'un job pour vérifier sa sortie et valider votre logique SELECT ou INSERT. Cette fonctionnalité améliore l'efficacité du développement et réduit les risques liés à la qualité des données.

   Remarque

   La fonctionnalité de débogage n'écrit pas de données dans la table de destination (sink).

   1. Dans le coin supérieur droit de l'éditeur SQL, cliquez sur Debug.

   2. Dans la boîte de dialogue Debug, sélectionnez un cluster de débogage et cliquez sur Next.

      Si aucun cluster de session n'est disponible, vous devez en créer un. Le cluster de session doit utiliser la même version du moteur que le brouillon SQL et être en cours d'exécution. Pour plus d'informations, consultez Étape 1 : Créer un cluster de session.

   3. Configurez les données de débogage et cliquez sur OK.

      Pour plus d'informations sur la configuration, consultez Étape 2 : Déboguer un job.

Étape 4 : Déployer le brouillon SQL

Dans le coin supérieur droit de l'éditeur SQL, cliquez sur Deploy. Dans la boîte de dialogue Deploy draft, configurez les paramètres selon vos besoins et cliquez sur OK.

Lors du déploiement du brouillon, vous pouvez sélectionner une Resource Queue ou un session cluster comme Deployment target. Le tableau suivant compare ces deux options.

Étape 5 : Démarrer le job et afficher les résultats

1. Dans le volet de navigation, cliquez sur O&M > Deployments.

2. Recherchez le job cible et cliquez sur Start dans la colonne Actions.

   Sélectionnez Initial Mode, puis cliquez sur Start. Le job est en cours d'exécution lorsque son statut passe à Running. Pour plus d'informations sur les paramètres de démarrage, consultez Démarrer un job.

3. Sur la page de détails Deployments, affichez les résultats du job Flink.

   1. Sur la page O&M > Deployments, cliquez sur le nom du job cible.

   2. Dans l'onglet Logs, cliquez sur le sous-onglet Running Task Managers. Dans la colonne Path, ID, cliquez sur un TaskManager.

   3. Cliquez sur l'onglet Logs et recherchez les journaux liés à PrintSinkOutputWriter.

      Si vous trouvez des entrées de journal contenant la chaîne Source: datagen_source → Calc → Sink: print_table, le flux de données est traité correctement.

(Facultatif) Étape 6 : Arrêter le job

Pour appliquer des modifications à un job (telles que des modifications de code, des mises à jour des paramètres WITH ou des changements de version), vous devez le redéployer, l'arrêter, puis le redémarrer. Un redémarrage est également nécessaire pour un démarrage sans état ou pour appliquer des modifications de configuration non dynamiques. Pour plus d'informations sur l'arrêt d'un job, consultez Arrêter un job.

1. Sur la page O&M > Deployments, recherchez le job cible et cliquez sur Cancel dans la colonne Actions.

2. Cliquez sur OK.

Documents connexes

• FAQ sur le développement et l'O&M des jobs

  • Développement de jobs

  • O&M des jobs

• Configurer les informations du job

  • Vous pouvez configurer les ressources avant de démarrer un job ou les modifier pour un déploiement en cours d'exécution. Deux modes de configuration des ressources sont pris en charge : basique (granularité grossière) et expert (granularité fine). Pour plus d'informations, consultez Configurer les ressources des jobs.

  • Vous pouvez configurer les niveaux de journalisation et spécifier différentes sorties pour chaque niveau. Pour plus d'informations, consultez Configurer les sorties des journaux de jobs.

• Flux de travail de développement pour d'autres types de jobs

  • Job Flink JAR

  • Job Flink Python

  • Ingestion de bases de données en temps réel

• Meilleures pratiques pour Flink

  • Construire un entrepôt de données en temps réel avec Hologres

  • Construire un lac de données en streaming avec Paimon et StarRocks