Realtime Compute for Apache Flink:Fonctions scalaires définies par l'utilisateur (UDSF) Python

Limites

Les contraintes suivantes s'appliquent au développement de fonctions définies par l'utilisateur (UDF) Python dans Realtime Compute for Apache Flink :

Méthodes de développement

Vous disposez de deux méthodes pour développer une UDSF Python :

• Empaquetez votre code Python, téléchargez le package, puis enregistrez-le en tant que fonction sur la plateforme.

• Déclarez la logique du code sous forme de fonction inline dans une instruction SQL.

Si la logique de votre UDSF est simple, privilégiez le développement sous forme de fonction inline.

Créer une UDSF

Les étapes suivantes prennent Windows comme environnement d'exemple. Flink fournit un référentiel d'exemples qui inclut des implémentations pour les UDSF, les fonctions d'agrégation définies par l'utilisateur (UDAF) et les fonctions table définies par l'utilisateur (UDTF).

1. Téléchargez et décompressez python_demo-master sur votre machine locale.

   Il s'agit d'un référentiel GitHub tiers. L'accès peut être lent ou intermittent.

2. Dans PyCharm, choisissez File > Open et ouvrez le répertoire python_demo-master décompressé.

3. Ouvrez udfs.py dans le chemin \python_demo-master\udx et définissez votre UDSF.

   from pyflink.table import DataTypes
   from pyflink.table.udf import udf
   
   @udf(result_type=DataTypes.STRING())
   def sub_string(s: str, begin: int, end: int):
       return s[begin:end]

   L'exemple sub_string extrait les caractères de la position begin à la position end dans la chaîne d'entrée.

4. Depuis le répertoire \python_demo-master, exécutez la commande suivante pour empaqueter le répertoire udx :

   zip -r python_demo.zip udx

   Lorsque python_demo.zip apparaît dans \python_demo-master\, le package est prêt.

Enregistrer une UDSF

Une fois le package créé, enregistrez l'UDSF dans la console Realtime Compute for Apache Flink. Pour connaître les étapes d'enregistrement, consultez Gérer les fonctions définies par l'utilisateur (UDF).

Utiliser une UDSF

Après avoir enregistré l'UDSF, utilisez-la dans un job Flink SQL.

1. Créez un brouillon à l'aide de Flink SQL. Pour plus de détails, consultez Présentation du développement de jobs. L'exemple suivant appelle ASI_UDSF (le nom enregistré de votre UDSF) pour extraire les caractères des positions 2 à 4 du champ a dans la table source :

   CREATE TEMPORARY TABLE ASI_UDSF_Source (
     a VARCHAR,
     b INT,
     c INT
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDSF_Sink (
     a VARCHAR
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDSF_Sink
   SELECT ASI_UDSF(a, 2, 4)
   FROM ASI_UDSF_Source;

2. Dans le volet de navigation de gauche de la console de développement, choisissez O&M > Deployments. Localisez le déploiement, puis cliquez sur Start dans la colonne Actions. Une fois le déploiement démarré, les caractères aux positions 2 à 4 du champ a dans ASI_UDSF_Source sont écrits dans ASI_UDSF_Sink.

Fonctions scalaires inline Python

Une fonction inline intègre son implémentation directement dans une instruction CREATE FUNCTION, ce qui permet de définir et d'enregistrer la fonction dans la même instruction SQL. L'exemple ci-dessous définit une fonction inline qui masque les adresses e-mail. Déclarez l'intégralité de la logique du code entre les délimiteurs $$.

CREATE TEMPORARY FUNCTION mask_email(email STRING)
RETURNS STRING AS $$
if email is None:
    return None

name, separator, domain = email.partition("@")
if not separator:
    return email

return name[:1] + "***@" + domain
$$ LANGUAGE PYTHON;

Fonctions définies par l'utilisateur asynchrones

Pour les UDF effectuant des opérations intensives en E/S, telles que l'accès à des bases de données externes ou des requêtes HTTP, utilisez des fonctions définies par l'utilisateur asynchrones. Une seule UDF asynchrone peut traiter plusieurs requêtes d'E/S simultanément, répartissant ainsi le temps d'attente entre les requêtes et améliorant le débit du job.

import asyncio

from pyflink.table import DataTypes
from pyflink.table.udf import AsyncScalarFunction, udf

# Method 1: Use a Python async function
@udf(result_type=DataTypes.STRING())
async def async_api_call(product_id: str) -> str:
    await asyncio.sleep(0.05)
    return f"product_{product_id}"

# Method 2: Subclass the asynchronous function class
class AsyncUserLookup(AsyncScalarFunction):
    def open(self, function_context):
        self.cache = {}

    async def eval(self, user_id: str) -> str:
        if user_id in self.cache:
            return self.cache[user_id]

        await asyncio.sleep(0.05)
        result = f"user_{user_id}"
        self.cache[user_id] = result
        return result

    def close(self):
        self.cache.clear()

async_user_lookup = udf(
    AsyncUserLookup(),
    input_types=[DataTypes.STRING()],
    result_type=DataTypes.STRING()
)