Realtime Compute for Apache Flink:Python

@udf(result_type=DataTypes.FLOAT())
def percentile(values: List[float], percentile: float):
    import numpy as np
    return np.percentile(values, percentile)

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

Préchargement des ressources

Préchargez les ressources lors de l'initialisation de la fonction pour éviter de les recharger à chaque appel de la méthode eval. Par exemple, chargez un modèle d'apprentissage profond volumineux une seule fois, puis effectuez des prédictions par lots.

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

class Predict(ScalarFunction):
    def open(self, function_context):
        import pickle

        with open("resources.zip/resources/model.pkl", "rb") as f:
            self.model = pickle.load(f)

    def eval(self, x):
        return self.model.predict(x)

predict = udf(Predict(), result_type=DataTypes.DOUBLE(), func_type="pandas")

Fonctions définies par l'utilisateur asynchrones

Pour les scénarios intensifs en E/S, tels que l'accès à des bases de données externes ou les appels de services HTTP, utilisez des fonctions définies par l'utilisateur asynchrones. Une seule instance de fonction peut gérer plusieurs requêtes simultanément, répartissant le temps d'attente sur plusieurs appels et améliorant considérablement le débit du job. Cette fonctionnalité est prise en charge uniquement dans les versions VVR 11.7 et ultérieures et s'applique uniquement aux fonctions scalaires définies par l'utilisateur (UDSF). Pour plus de détails, consultez la rubrique Fonctions définies par l'utilisateur asynchrones.

Utilisation de la bibliothèque Pandas

Outre les fonctions définies par l'utilisateur Python standard, Realtime Compute for Apache Flink prend en charge les fonctions définies par l'utilisateur Pandas. Ces fonctions acceptent des structures de données Pandas telles que pandas.Series et pandas.DataFrame en entrée, ce qui permet d'exploiter des bibliothèques hautes performances comme Pandas et NumPy. Pour plus d'informations, consultez la documentation Vectorized User-defined Functions.

Paramètres

Les performances d'une fonction définie par l'utilisateur Python dépendent largement de son implémentation. En cas de problèmes de performances, commencez par optimiser la logique de la fonction. Les paramètres suivants influent également sur les performances.