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

Precarga de recursos

Precargue los recursos durante la inicialización de la función para evitar recargarlos en cada llamada al método eval. Por ejemplo, cargue un modelo grande de aprendizaje profundo una sola vez y luego ejecute predicciones por lotes sobre él.

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")

Funciones definidas por el usuario asíncronas

Para escenarios con uso intensivo de E/S, como el acceso a bases de datos externas o llamadas a servicios HTTP, utilice funciones definidas por el usuario asíncronas. Una sola instancia de función puede manejar varias solicitudes simultáneamente; esto distribuye el tiempo de espera entre múltiples llamadas y mejora significativamente el throughput del trabajo. Esta función solo es compatible con VVR 11.7 y versiones posteriores, y aplica únicamente a funciones escalares definidas por el usuario (UDSF). Para más detalles, consulte Asynchronous user-defined functions.

Uso de la biblioteca Pandas

Además de las funciones definidas por el usuario estándar de Python, Realtime Compute for Apache Flink admite funciones definidas por el usuario de Pandas. Estas funciones aceptan estructuras de datos de Pandas, como pandas.Series y pandas.DataFrame, como entrada, lo que le permite aprovechar bibliotecas de alto rendimiento como Pandas y NumPy. Para obtener más información, consulte Funciones definidas por el usuario vectorizadas.

Parámetros

El rendimiento de una función definida por el usuario de Python depende en gran medida de su implementación. Si experimenta problemas de rendimiento, optimice primero la lógica de la función. Los siguientes parámetros también afectan el rendimiento.