Realtime Compute for Apache Flink:Python UDAF'leri

Sınırlamalar

• Apache Flink 1.12 veya sonraki bir sürüm gerekir.

• Python, Realtime Compute for Apache Flink çalışma alanında önceden yüklenmiştir. Kodunuzu önceden yüklenmiş Python sürümünde yazın.

  Ververica Runtime (VVR) 8.0.11'den önceki sürümler için Python 3.7.9 önceden yüklenmiştir. VVR 8.0.11 veya sonraki sürümler için Python 3.9.21 önceden yüklenmiştir. VVR 8.0.11 veya sonraki bir sürüme yükselttikten sonra, daha önceki bir VVR sürümü üzerine oluşturulan PyFlink işlerini yeniden test edin, yeniden dağıtın ve yeniden çalıştırın.

• Çalışma zamanı ortamında JDK 8 ve JDK 11 desteklenir. Python dağıtımınız üçüncü taraf bir JAR dosyasına bağımlıysa JAR dosyasının JDK 8 veya JDK 11 ile uyumlu olduğundan emin olun.

• Yalnızca açık kaynak Scala 2.11 desteklenir. Python dağıtımınız üçüncü taraf bir JAR dosyasına bağımlıysa JAR dosyasının Scala 2.11 ile uyumlu olduğundan emin olun.

Çalışma mantığı

Bir UDAF, girdi satırları arasındaki ara toplama durumunu izlemek için bir akümülatör kullanır. Akümülatör, grup başına bir kez oluşturulur ve tüm girdiler işlenene kadar satır satır güncellenir.

Her grup için yürütme sırası şu şekildedir:

1. create_accumulator() — başlangıç durumunu tutmak için boş bir akümülatör oluşturur.

2. accumulate(accumulator, ...) — akümülatörü güncellemek üzere her girdi satırı için bir kez çağrılır.

3. get_value(accumulator) — gruptaki tüm satırlar işlendikten sonra nihai sonucu döndürmek üzere çağrılır.

UDAF oluşturma

Flink, UDAF'ler, kullanıcı tanımlı işlevler (UDF'ler) ve kullanıcı tanımlı tablo işlevleri (UDTF'ler) dahil olmak üzere kullanıcı tanımlı uzantılar (UDX'ler) için örnek kod sağlar. Aşağıdaki adımlar Windows ortamını kullanır.

1. python_demo-master paketini makinenize indirin ve arşivden çıkarın.

2. PyCharm'da File > Open seçeneğini belirleyin ve arşivden çıkarılmış python_demo-master dizinini açın.

3. \python_demo-master\udx\udfs.py dosyasını açın ve iş mantığınıza uyacak şekilde düzenleyin. Aşağıdaki örnekte, mevcut ve geçmiş veriler üzerinden ağırlıklı ortalamayı hesaplayan weighted_avg tanımlanmıştır.

   from pyflink.common import Row
   from pyflink.table import AggregateFunction, DataTypes
   from pyflink.table.udf import udaf
   
   class WeightedAvg(AggregateFunction):
   
       def create_accumulator(self):
           # Row(sum, count)
           return Row(0, 0)
   
       def get_value(self, accumulator: Row) -> float:
           if accumulator[1] == 0:
               return 0
           else:
               return accumulator[0] / accumulator[1]
   
       def accumulate(self, accumulator: Row, value, weight):
           accumulator[0] += value * weight
           accumulator[1] += weight
   
       def retract(self, accumulator: Row, value, weight):
           accumulator[0] -= value * weight
           accumulator[1] -= weight
   
   weighted_avg = udaf(f=WeightedAvg(),
                       result_type=DataTypes.DOUBLE(),
                       accumulator_type=DataTypes.ROW([
                           DataTypes.FIELD("f0", DataTypes.BIGINT()),
                           DataTypes.FIELD("f1", DataTypes.BIGINT())]))

4. \python_demo-master\ dizininden udx klasörünü paketleyin:

   zip -r python_demo.zip udx

   python_demo.zip dosyası \python_demo-master\ dizininde göründüğünde UDAF oluşturulmuş olur.

UDAF kaydetme

Kayıt adımları için bkz. Kullanıcı tanımlı işlevleri (UDF'ler) yönetme.

UDAF kullanma

1. Bir Flink SQL taslağı oluşturun. Ayrıntılar için bkz. İş geliştirme genel bakışı. Aşağıdaki örnekte, ağırlık olarak b alanı kullanılarak ASI_UDAF_Source içindeki a alanının ağırlıklı ortalaması hesaplanır.

   CREATE TEMPORARY TABLE ASI_UDAF_Source (
     a   BIGINT,
     b   BIGINT
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDAF_Sink (
     avg_value  DOUBLE
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDAF_Sink
   SELECT weighted_avg(a, b)
   FROM ASI_UDAF_Source;

2. Realtime Compute for Apache Flink geliştirme konsolunun sol gezinme bölmesinde O&M > Deployments seçeneğini belirleyin. Hedef dağıtımı bulun ve Actions sütunundaki Start seçeneğine tıklayın. Dağıtım başladıktan sonra, ağırlık olarak b alanı kullanılarak a alanının ağırlıklı ortalaması ASI_UDAF_Sink içindeki her satıra yazılır.