UDTF de Python - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Este tema describe cómo crear, registrar y usar una función tabular definida por el usuario (UDTF) de Python en Realtime Compute for Apache Flink.

Descripción general

Una UDTF acepta cero, uno o varios valores escalares como parámetros de entrada. Estos parámetros pueden tener longitud variable. Las UDTF devuelven cualquier cantidad de filas en lugar de un solo valor, y dichas filas pueden incluir una o más columnas. Cada vez que se invoca una UDTF, se obtienen varias filas o columnas. Las UDTF son similares a las funciones escalares definidas por el usuario (UDF), pero difieren en los resultados que devuelven.

Límites

Al desarrollar funciones definidas por el usuario (UDF) de Python en Realtime Compute for Apache Flink, aplican las siguientes restricciones:

Restricción	Requisito
Versión de Apache Flink	1.12 y posteriores
Versión de Python	Preinstalada en cada espacio de trabajo. VVR anterior a 8.0.11: Python 3.7.9. VVR 8.0.11 y posteriores: Python 3.9.21.
Versión de JDK	JDK 8 y JDK 11. Los paquetes JAR de terceros deben ser compatibles con JDK 8 o JDK 11.
Versión de Scala	Solo Scala 2.11 de código abierto. Los paquetes JAR de terceros deben ser compatibles con Scala 2.11.
Funciones en línea	Compatible solo con VVR 11.9-preview1 y versiones posteriores.

Importante

Después de actualizar a VVR 8.0.11 o posterior, pruebe, despliegue y ejecute nuevamente sus borradores existentes de PyFlink para confirmar su compatibilidad.

Crear una UDTF

Nota

Flink proporciona código de ejemplo de extensiones definidas por el usuario (UDX) de Python para facilitar el desarrollo de UDX. Este código incluye la implementación de UDF de Python, funciones de agregación definidas por el usuario (UDAF) de Python y UDTF de Python. Esta sección describe cómo crear una UDTF en el sistema operativo Windows.

Importante

El nombre registrado de una función personalizada solo puede contener letras minúsculas, dígitos y guiones (-). No se admiten guiones bajos (_) ni otros caracteres especiales.

El nombre de la función no puede coincidir con ninguna función integrada. El entorno administrado de Flink tiene pre-registradas algunas funciones integradas como split. Si una función personalizada tiene el mismo nombre que una función integrada, el registro fallará o se producirá un error de validación SQL. Se recomienda utilizar un prefijo personalizado para los nombres de las funciones, como my-split.

Descargue y descomprima python_demo-master en su computadora local.
En PyCharm, seleccione File > Open y abra el directorio descomprimido python_demo-master.
Haga doble clic en el archivo udtfs.py ubicado en el directorio \python_demo-master\udx. A continuación, modifique el contenido del archivo según sus requisitos empresariales.
En este ejemplo, my_split define el código que separa una fila de cadena en varias columnas de cadenas mediante barras verticales (|).
```
from pyflink.table import DataTypes
from pyflink.table.udf import udtf

@udtf(result_types=[DataTypes.STRING(), DataTypes.STRING()])
def my_split(s: str):
    splits = s.split("|")
    yield splits[0], splits[1]
```
Vaya al directorio \python_demo donde se encuentra la carpeta udx y ejecute el siguiente comando para empaquetar los archivos del directorio:
```
zip -r python_demo.zip udx
```
Si el paquete python_demo.zip aparece en el directorio \python_demo\, significa que la UDTF se ha desarrollado correctamente.

Registrar una UDTF

Para obtener más información sobre cómo registrar una UDTF, consulte Manage UDFs.

Usar una UDTF

Una vez registrada la UDTF, siga estos pasos para utilizarla:

Use Flink SQL para crear un borrador. Para más información, consulte Job development overview.

Después de concatenar la cadena "aa" y el campo message de cada fila en la tabla ASI_UDTF_Source mediante barras verticales (|), las cadenas resultantes se separan en varias columnas con dicho carácter. El siguiente código muestra un ejemplo:

CREATE TEMPORARY TABLE ASI_UDTF_Source (
  `message`  VARCHAR
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE ASI_UDTF_Sink (
  name  VARCHAR,
  place  VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO ASI_UDTF_Sink
SELECT name,place
FROM ASI_UDTF_Source,lateral table(my_split(concat_ws('|', `message`, 'aa'))) as T(name,place);

En el panel de navegación izquierdo de la consola de desarrollo de Realtime Compute for Apache Flink, seleccione O&M > Deployments. En la página Deployments, localice el despliegue deseado y haga clic en Start dentro de la columna Actions.

Al iniciar el despliegue, se insertarán dos columnas de datos en la tabla ASI_UDTF_Sink. Estas columnas contienen las cadenas concatenadas separadas por barras verticales (|).