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

Kaynakların önceden yüklenmesi

eval yöntemine yapılan her çağrıda kaynakların yeniden yüklenmesini önlemek için işlev başlatma sırasında kaynakları önceden yükleyin. Örneğin, büyük bir derin öğrenme modelini bir kez yükleyip ardından bu modele karşı toplu iş tahminleri çalıştırabilirsiniz.

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

Eşzamansız kullanıcı tanımlı işlevler

Harici veritabanlarına erişme veya HTTP hizmeti çağrıları yapma gibi G/Ç yoğun senaryolar için eşzamansız kullanıcı tanımlı işlevleri kullanın. Tek bir işlev örneği, birden çok isteği aynı anda işleyebilir; bekleme süresini birkaç çağrıya dağıtarak iş aktarım hızını önemli ölçüde artırır. Bu özellik yalnızca VVR 11.7 ve sonraki sürümlerde desteklenir ve yalnızca kullanıcı tanımlı skaler işlevler (UDSF) için geçerlidir. Ayrıntılar için bkz. Eşzamansız kullanıcı tanımlı işlevler.

Pandas kitaplığını kullanma

Standart Python kullanıcı tanımlı işlevlerine ek olarak Realtime Compute for Apache Flink, Pandas kullanıcı tanımlı işlevleri de destekler. Bu işlevler, pandas.Series ve pandas.DataFrame gibi Pandas veri yapılarını giriş olarak kabul eder; böylece Pandas ve NumPy gibi yüksek performanslı kitaplıklardan yararlanabilirsiniz. Daha fazla bilgi için bkz. Vektörleştirilmiş Kullanıcı Tanımlı İşlevler.

Parametreler

Bir Python kullanıcı tanımlı işlevinin performansı büyük ölçüde uygulamasına bağlıdır. Performans sorunlarıyla karşılaşırsanız önce işlev mantığını optimize edin. Aşağıdaki parametreler de performansı etkiler.