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é-carregamento de recursos

Carregue os recursos durante a inicialização da função para evitar recarregamentos a cada chamada do método eval. Essa abordagem permite, por exemplo, carregar um modelo grande de deep learning apenas uma vez e executar previsões em lote posteriormente.

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

Funções definidas pelo usuário assíncronas

Em cenários com uso intensivo de I/O, como acesso a bancos de dados externos ou chamadas a serviços HTTP, use funções definidas pelo usuário assíncronas. Uma única instância da função processa múltiplas requisições simultaneamente, distribuindo o tempo de espera entre várias chamadas e aumentando significativamente o throughput do job. Esse recurso está disponível apenas no VVR 11.7 e versões posteriores, aplicando-se exclusivamente a funções escalares definidas pelo usuário (UDSFs). Para obter detalhes, consulte Funções definidas pelo usuário assíncronas.

Uso da biblioteca Pandas

Além das funções definidas pelo usuário padrão em Python, o Realtime Compute for Apache Flink oferece suporte a UDFs baseadas em Pandas. Essas funções aceitam estruturas de dados do Pandas, como pandas.Series e pandas.DataFrame, como entrada, permitindo o uso de bibliotecas de alto desempenho como Pandas e NumPy. Para obter mais informações, consulte Funções Definidas pelo Usuário Vetorizadas.

Parâmetros

O desempenho de uma função definida pelo usuário em Python depende principalmente da implementação. Em caso de problemas de performance, otimize primeiro a lógica da função. Os parâmetros listados abaixo também influenciam o desempenho.