Realtime Compute for Apache Flink:UDAFs de Python

Limitaciones

• Se requiere Apache Flink 1.12 o posterior.

• Python viene preinstalado en el espacio de trabajo de Realtime Compute for Apache Flink. Escriba su código con la versión de Python preinstalada.

  Python 3.7.9 está preinstalado para versiones de Ververica Runtime (VVR) anteriores a la 8.0.11. Python 3.9.21 está preinstalado para VVR 8.0.11 o posterior. Después de actualizar a VVR 8.0.11 o posterior, vuelva a probar, desplegar y ejecutar cualquier trabajo de PyFlink creado con una versión anterior de VVR.

• El entorno de ejecución es compatible con JDK 8 y JDK 11. Si su despliegue de Python depende de un archivo JAR de terceros, asegúrese de que dicho archivo sea compatible con JDK 8 o JDK 11.

• Solo se admite Scala 2.11 de código abierto. Si su despliegue de Python depende de un archivo JAR de terceros, asegúrese de que dicho archivo sea compatible con Scala 2.11.

Funcionamiento

Una UDAF utiliza un acumulador para rastrear el estado intermedio de agregación a través de las filas de entrada. El acumulador se crea una vez por grupo y se actualiza fila por fila hasta procesar todas las entradas.

El orden de ejecución para cada grupo es el siguiente:

1. create_accumulator() — crea un acumulador vacío para almacenar el estado inicial.

2. accumulate(accumulator, ...) — se llama una vez por cada fila de entrada para actualizar el acumulador.

3. get_value(accumulator) — se llama después de procesar todas las filas del grupo para devolver el resultado final.

Crear una UDAF

Flink proporciona código de ejemplo para extensiones definidas por el usuario (UDX), incluidas UDAFs, funciones definidas por el usuario (UDF) y funciones de tabla definidas por el usuario (UDTF). Los pasos siguientes utilizan un entorno Windows.

1. Descargue y descomprima el paquete python_demo-master en su computadora.

2. En PyCharm, elija File > Open y abra el directorio descomprimido python_demo-master.

3. Abra \python_demo-master\udx\udfs.py y edítelo para ajustarlo a su lógica de negocio. El siguiente ejemplo define weighted_avg, que calcula un promedio ponderado sobre datos actuales e históricos.

   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. Desde el directorio \python_demo-master\, empaquete la carpeta udx:

   zip -r python_demo.zip udx

   La UDAF se crea cuando aparece python_demo.zip en el directorio \python_demo-master\.

Registrar una UDAF

Para consultar los pasos de registro, véase Manage user-defined functions (UDFs).

Usar una UDAF

1. Cree un borrador de Flink SQL. Para más detalles, consulte Job development overview. El siguiente ejemplo calcula el promedio ponderado del campo a en ASI_UDAF_Source y usa el campo b como peso.

   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. En el panel de navegación izquierdo de la consola de desarrollo de Realtime Compute for Apache Flink, elija O&M > Deployments. Localice el despliegue objetivo y haga clic en Start en la columna Actions. Una vez iniciado el despliegue, el promedio ponderado del campo a (con el campo b como peso) se escribe en cada fila de ASI_UDAF_Sink.