Realtime Compute for Apache Flink:Développer des jobs Python

Choix entre Table API/SQL et DataStream API

PyFlink propose deux approches de développement : Table API/SQL et DataStream API. Table API/SQL est recommandé pour les raisons suivantes :

• Meilleures performances : Le plan d'exécution optimisé de Table API/SQL s'exécute entièrement au sein de la JVM. À l'inverse, DataStream API nécessite une sérialisation et une désérialisation ligne par ligne entre la JVM et le processus Python, ce qui engendre une surcharge significative.

• Fonctionnalités plus complètes : Table API/SQL offre une prise en charge étendue des connecteurs, formats de données et fonctions de fenêtrage, tout en partageant l'écosystème de connecteurs des jobs SQL.

• Recommandation communautaire : La communauté Apache Flink privilégie Table API/SQL comme axe principal de développement pour PyFlink.

Réservez DataStream API aux scénarios nécessitant une logique personnalisée complexe impossible à exprimer en SQL.

Références de développement

Consultez les documents suivants pour développer votre code métier Flink en local. Une fois le développement terminé, importez le code sur la console de développement Flink et déployez le job.

• Pour le développement de code métier Apache Flink V1.20, reportez-vous au Guide de développement Flink Python API.

• Pour résoudre les problèmes rencontrés lors du codage Apache Flink, consultez la FAQ.

Structure du projet

La structure de projet recommandée pour les jobs Python est la suivante :

my-flink-python-project/
├── my_job.py                # Main job file.
├── udfs.py                  # User-defined functions (optional)
├── requirements.txt         # Third-party Python dependencies (optional)
└── config.properties        # Configuration file (optional)

Gestion des dépendances

Pour savoir comment utiliser des environnements virtuels Python personnalisés, des packages Python tiers, des fichiers JAR et des fichiers de données dans vos jobs Python, consultez Utiliser des dépendances Python.

Fonctions définies par l'utilisateur (UDF)

Voici un exemple de développement d'une UDSF Python permettant de masquer des données sensibles sous forme de chaîne :

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())def mask_phone(phone: str):"""Phone number masking: retains the first 3 and last 4 digits, replacing the middle with ****"""if phone is None or len(phone) != 11:return phone.
    return phone[:3] + '****' + phone[7:]

Utilisez cette UDF dans un job SQL :

CREATE TEMPORARY FUNCTION mask_phone AS 'udfs.mask_phone' LANGUAGE PYTHON;INSERT INTO sink_table
SELECT name, mask_phone(phone) AS masked_phone
FROM source_table;

Pour enregistrer, mettre à jour ou supprimer des UDF, reportez-vous à Gérer les UDF.

Utilisation des connecteurs

Pour obtenir la liste des connecteurs pris en charge par Flink, consultez Connecteurs pris en charge. Pour utiliser un connecteur, procédez comme suit :

1. Connectez-vous à la console Realtime Compute.

2. Cliquez sur Console dans la colonne Actions de l'espace de travail cible.

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

4. Cliquez sur Upload Artifact et sélectionnez le package Python du connecteur souhaité pour le charger.

   Vous pouvez charger un connecteur développé en interne ou fourni par Flink. Pour télécharger les packages Python officiels des connecteurs Flink, consultez la liste des connecteurs.

5. Sur la page O&M > Deployments, cliquez sur Create Deployment > Python Deployment. Sélectionnez le package Python du connecteur cible dans le champ Additional Dependencies, configurez les autres paramètres, puis déployez le job.

6. Cliquez sur le nom du job déployé. Sous l'onglet Configuration, dans la section Parameters, cliquez sur Edit. Dans Other Configuration, ajoutez les informations d'emplacement des packages Python du connecteur.

   Si votre job dépend de plusieurs packages Python de connecteurs, par exemple deux packages nommés connector-1.jar et connector-2.jar, utilisez la configuration suivante.

   pipeline.classpaths: 'file:///flink/usrlib/connector-1.jar;file:///flink/usrlib/connector-2.jar'

7. Pour utiliser des connecteurs intégrés, des formats de données et des catalogues (uniquement pour VVR 11.2 et ultérieur), ajoutez la configuration dans la section Running Parameters sous Additional Configuration du job. Exemple :

   ## Multiple connectors used.
   pipeline.used-builtin-connectors: kafka;sls
   ## Multiple data formats for data transmission.
   pipeline.used-builtin-formats: avro;parquet
   ## Multiple previously created catalogs used.
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

Pour plus de détails sur l'utilisation des connecteurs, consultez Exemple de code complet.

Débogage des jobs

import logging

@udf(result_type=DataTypes.BIGINT())
def add(i, j):
  logging.info("hello world")
  return i + j

Débogage local

Realtime Compute for Apache Flink n'ayant pas accès à Internet par défaut, votre code pourrait ne pas pouvoir se connecter directement aux sources de données en ligne pour des tests locaux. Nous recommandons les approches suivantes pour le débogage local :

• Tests unitaires : Réalisez des tests unitaires indépendants sur les UDF afin de valider la logique des fonctions.

• Exécution locale : Simulez les entrées à l'aide de sources de données locales (fichiers ou données en mémoire) et exécutez le job localement pour vérifier la logique de traitement. Exemple :

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()# Use a local data source for testing.
  ds = env.from_collection([('Alice', 1), ('Bob', 2), ('Alice', 3)])
  ds.key_by(lambda x: x[0]).sum(1).print()
  env.execute("local_test")

• Débogage distant : Pour déboguer en vous connectant à des sources de données en ligne, consultez Exécuter et déboguer localement des jobs avec connecteurs.

Déploiement des jobs

Une fois le développement du job Python terminé, chargez-le sur la console Realtime Compute pour le déployer. Procédez comme suit :

1. Connectez-vous à la console Realtime Compute et accédez à l'espace de travail cible.

2. Dans le volet de navigation de gauche, cliquez sur Artifacts et chargez le fichier du job Python (.py ou .zip). Chargez également les dépendances tierces ou fichiers de configuration si nécessaire.

3. Sur la page O&M > Deployments, cliquez sur Create Deployment > Python Deployment et renseignez les informations de déploiement.

   Parameter	Description
   Python File Path	Sélectionnez le fichier du job Python précédemment chargé.
   Entry Module	Ce champ est facultatif pour un fichier .py. Pour un fichier .zip, spécifiez le nom du module d'entrée, par exemple my_job.
   Additional Dependencies	Sélectionnez les fichiers JAR de connecteurs ou de configuration le cas échéant.
   Python Libraries	Sélectionnez les packages Python tiers (.whl ou .zip) si nécessaire.
   Python Archives	Sélectionnez un environnement virtuel Python personnalisé (.zip) le cas échéant.

4. Cliquez sur Deploy.

Pour plus d'informations sur les paramètres de déploiement, consultez Déployer un job.

import logging
import sys

from pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

logging.basicConfig(stream=sys.stdout, level=logging.INFO)

def kafka_to_mysql():
    # Create the execution environment.
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)

    # Create the Kafka source table.
    t_env.execute_sql("""
        CREATE TABLE kafka_source (
            `id` INT,
            `name` STRING,
            `score` INT,
            `event_time` TIMESTAMP(3),
            WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
        ) WITH (
            'connector' = 'kafka',
            'topic' = 'student_topic',
            'properties.bootstrap.servers' = 'your-kafka-broker:9092',
            'properties.group.id' = 'my-group',
            'scan.startup.mode' = 'latest-offset',
            'format' = 'json'
        )
    """)

    # Create the MySQL sink table.
    t_env.execute_sql("""
        CREATE TABLE mysql_sink (
            `id` INT,
            `name` STRING,
            `score` INT,
            PRIMARY KEY (id) NOT ENFORCED
        ) WITH (
            'connector' = 'jdbc',
            'url' = 'jdbc:mysql://your-mysql-host:3306/my_database',
            'table-name' = 'student',
            'username' = 'your_username',
            'password' = 'your_password'
        )
    """)

    # Filter records with scores greater than 60 and write them to MySQL.
    t_env.execute_sql("""
        INSERT INTO mysql_sink
        SELECT id, name, score
        FROM kafka_source
        WHERE score >= 60
    """)

if __name__ == '__main__':
    kafka_to_mysql()