Realtime Compute for Apache Flink:UDTF Python

Développer une UDTF

Remarque

Flink fournit un exemple de projet d'extensions utilisateur définies (UDX) en Python pour vous aider à développer des UDX. L'exemple inclut les implémentations de fonctions scalaires utilisateur définies, de fonctions d'agrégation utilisateur définies (UDAF) et d'UDTF. Cet exemple utilise le système d'exploitation Windows pour expliquer comment développer une UDTF.

Important

• Un nom de UDF enregistré ne peut contenir que des lettres minuscules, des chiffres et des traits d'union (-). Les traits de soulignement (\_) et autres caractères spéciaux ne sont pas pris en charge.

• Le nom de la fonction ne doit pas être identique à celui d'une fonction intégrée. L'environnement géré de Realtime Compute for Apache Flink comprend certaines fonctions intégrées préinstallées, telles que split. Si une fonction personnalisée porte le même nom qu'une fonction intégrée, l'enregistrement échoue ou la validation SQL signale une erreur. Utilisez un préfixe personnalisé pour le nom de la fonction, par exemple my-split.

1. Téléchargez et décompressez le package d'exemple python\_demo-master sur votre ordinateur.

2. Dans PyCharm, choisissez file > open pour ouvrir le dossier python\_demo-master décompressé.

3. Double-cliquez sur \python\_demo-master\udx\udtfs.py et modifiez le contenu du fichier udtfs.py en fonction de vos besoins métier.

Dans cet exemple, my\_split divise la chaîne de chaque ligne en plusieurs colonnes à l'aide de barres verticales (|).

    from pyflink.table import DataTypes
    from pyflink.table.udf import udtf
    
    @udtf(result_types=[DataTypes.STRING(), DataTypes.STRING()])
    def my_split(s: str):
        splits = s.split("|")
        yield splits[0], splits[1]
        

1. Dans le répertoire contenant le dossier udx du package téléchargé (\python\_demo), exécutez la commande suivante pour empaqueter les fichiers.

    zip -r python_demo.zip udx
        

Le package python\_demo.zip est généré dans le répertoire \python\_demo\, ce qui indique que l'UDTF est développée.

Enregistrer une UDTF

Pour plus d'informations sur l'enregistrement d'une UDTF, consultez Gérer les UDF.

Utiliser une UDTF

Une fois l'UDTF enregistrée, vous pouvez l'utiliser. Procédez comme suit :

1. Développez un job Flink SQL. Pour plus d'informations, consultez Présentation du développement de jobs.

La valeur du champ message de chaque ligne de la table ASI\_UDTF\_Source est concaténée avec la chaîne aa à l'aide d'une barre verticale (|), puis divisée en plusieurs colonnes à l'aide de barres verticales (|). Le code suivant illustre cette opération :

    CREATE TEMPORARY TABLE ASI_UDTF_Source (
      `message`  VARCHAR
    ) WITH (
      'connector'='datagen'
    );
    
    CREATE TEMPORARY TABLE ASI_UDTF_Sink (
      name  VARCHAR,
      place  VARCHAR
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO ASI_UDTF_Sink
    SELECT name,place
    FROM ASI_UDTF_Source,lateral table(my_split(concat_ws('|', `message`, 'aa'))) as T(name,place);
        

1. Sur la page O&M > Deployments, cliquez sur Start dans la colonne Actions du job cible.

Une fois le job démarré, la table ASI\_UDTF\_Sink contient deux colonnes. Les données sont générées en concaténant les champs message et aa de chaque ligne de la table ASI\_UDTF\_Source avec une barre verticale (|), puis en divisant le résultat à l'aide de barres verticales (|).

CREATE TEMPORARY FUNCTION parse_order_items(items_json STRING)
RETURNS TABLE (
    sku STRING,
    quantity INT
)
HANDLER 'ParseOrderItems'
AS $$
import json

class ParseOrderItems:
    def eval(self, items_json):
        if not items_json:
            return

        try:
            items = json.loads(items_json)
        except (TypeError, ValueError):
            return

        if not isinstance(items, list):
            return

        for item in items:
            if not isinstance(item, dict):
                continue

            sku = str(item.get("sku", "")).strip().upper()

            try:
                quantity = int(item.get("quantity"))
            except (TypeError, ValueError):
                continue

            if sku and quantity > 0:
                yield sku, quantity
$$
LANGUAGE PYTHON;