Realtime Compute for Apache Flink:Python user-defined scalar functions (UDSFs)

Límites

A continuación se describen las restricciones aplicables al desarrollar funciones definidas por el usuario (UDF) de Python en Realtime Compute for Apache Flink:

Métodos de desarrollo

Puede desarrollar una UDSF de Python de cualquiera de las siguientes maneras:

• Empaquete su código de Python, cargue el paquete y regístrelo como una función en la plataforma.

• Declare la lógica del código como una función en línea dentro de una sentencia SQL.

Si la lógica de su UDSF es sencilla, considere desarrollarla como una función en línea.

Crear una UDSF

Los siguientes pasos utilizan Windows como entorno de ejemplo. Flink proporciona un repositorio de ejemplos que incluye implementaciones de UDSF, funciones de agregación definidas por el usuario (UDAF) y funciones con valor de tabla definidas por el usuario (UDTF).

1. Descargue y descomprima python_demo-master en su computadora local.

   Este es un repositorio de GitHub de terceros. El acceso puede ser lento o intermitente.

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

3. Abra udfs.py en la ruta \python_demo-master\udx y defina su UDSF.

   from pyflink.table import DataTypes
   from pyflink.table.udf import udf
   
   @udf(result_type=DataTypes.STRING())
   def sub_string(s: str, begin: int, end: int):
       return s[begin:end]

   El ejemplo sub_string extrae caracteres desde la posición begin hasta la posición end en la cadena de entrada.

4. Desde el directorio \python_demo-master, ejecute el siguiente comando para empaquetar el directorio udx:

   zip -r python_demo.zip udx

   Cuando aparezca python_demo.zip en \python_demo-master\, el paquete está listo.

Registrar una UDSF

Después de crear el paquete, registre la UDSF en la consola de Realtime Compute for Apache Flink. Para consultar los pasos de registro, véase Manage user-defined functions (UDFs).

Utilizar una UDSF

Una vez registrada la UDSF, utilícela en un trabajo de Flink SQL.

1. Cree un borrador mediante Flink SQL. Para más detalles, consulte Job development overview. El siguiente ejemplo llama a ASI_UDSF (el nombre registrado de su UDSF) para extraer los caracteres de las posiciones 2 a 4 del campo a en la tabla de origen:

   CREATE TEMPORARY TABLE ASI_UDSF_Source (
     a VARCHAR,
     b INT,
     c INT
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDSF_Sink (
     a VARCHAR
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDSF_Sink
   SELECT ASI_UDSF(a, 2, 4)
   FROM ASI_UDSF_Source;

2. En el panel de navegación izquierdo de la consola de desarrollo, seleccione O&M > Deployments. Localice el despliegue y haga clic en Start en la columna Actions. Una vez iniciado el despliegue, los caracteres en las posiciones 2–4 del campo a en ASI_UDSF_Source se escriben en ASI_UDSF_Sink.

Funciones escalares en línea de Python

Una función en línea integra su implementación directamente en una sentencia CREATE FUNCTION, por lo que la función se define y registra dentro de la misma sentencia SQL. El siguiente ejemplo define una función en línea que oculta parcialmente direcciones de correo electrónico. Declare la lógica completa del código entre los delimitadores $$.

CREATE TEMPORARY FUNCTION mask_email(email STRING)
RETURNS STRING AS $$
if email is None:
    return None

name, separator, domain = email.partition("@")
if not separator:
    return email

return name[:1] + "***@" + domain
$$ LANGUAGE PYTHON;

Funciones definidas por el usuario asíncronas

Para las UDF que realizan operaciones intensivas en E/S, como el acceso a bases de datos externas o solicitudes HTTP, utilice funciones definidas por el usuario asíncronas. Una sola UDF asíncrona puede gestionar múltiples solicitudes de E/S simultáneamente, lo que distribuye el tiempo de espera entre las solicitudes y mejora el throughput del trabajo.

import asyncio

from pyflink.table import DataTypes
from pyflink.table.udf import AsyncScalarFunction, udf

# Method 1: Use a Python async function
@udf(result_type=DataTypes.STRING())
async def async_api_call(product_id: str) -> str:
    await asyncio.sleep(0.05)
    return f"product_{product_id}"

# Method 2: Subclass the asynchronous function class
class AsyncUserLookup(AsyncScalarFunction):
    def open(self, function_context):
        self.cache = {}

    async def eval(self, user_id: str) -> str:
        if user_id in self.cache:
            return self.cache[user_id]

        await asyncio.sleep(0.05)
        result = f"user_{user_id}"
        self.cache[user_id] = result
        return result

    def close(self):
        self.cache.clear()

async_user_lookup = udf(
    AsyncUserLookup(),
    input_types=[DataTypes.STRING()],
    result_type=DataTypes.STRING()
)