Realtime Compute for Apache Flink:Flink CDC UDFs

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-cdc-common</artifactId>
    <version>${Apache Flink CDC version}</version>
    <scope>provided</scope>
</dependency>

public class AddOneFunctionClass implements UserDefinedFunction {

    public Object eval(Integer num) {
        return num + 1;
    }

    @Override
    public DataType getReturnType() {
        // The return type of eval is ambiguous (Object).
        // Use getReturnType to declare the exact CDC type.
        return DataTypes.INT();
    }

    @Override
    public void open() throws Exception {
        // Initialize resources here, for example, open a database connection.
    }

    @Override
    public void close() throws Exception {
        // Release resources here, for example, close a database connection.
    }
}

Mapeo de tipos

La siguiente tabla muestra la correspondencia entre los tipos de columna de CDC y los tipos de Java utilizados en los parámetros del método eval, los valores de retorno y getReturnType:

La ruta de la clase VARIANT difiere de la ruta de la clase Variant en Flink SQL.

pipeline:
  user-defined-function:
    - name: inc
      classpath: org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass
    - name: format
      classpath: org.apache.flink.cdc.udf.examples.java.FormatFunctionClass

Reglas de sintaxis

Una UDF de Python debe cumplir los siguientes requisitos:

• Se debe definir una función de nivel superior llamada eval en python-code.

• La función eval debe declarar una anotación de tipo de retorno.

• El número de parámetros de la función eval debe coincidir con el número de argumentos pasados al invocar la UDF.

• Las anotaciones de tipo de parámetro no son obligatorias, pero se recomienda mantenerlas para mejorar la legibilidad del código.

• Actualmente, las UDF de Python solo admiten funciones escalares. Cada invocación devuelve un único valor.

• Los valores NULL de SQL en los datos de entrada se convierten a None en Python.

• Una función de Python puede devolver None. En ese caso, la salida será NULL en SQL.

Por ejemplo, la siguiente UDF de Python elimina los espacios en blanco de la cadena de entrada y la convierte a minúsculas:

def eval(value: str) -> str:
    return value.strip().lower()

Puede definir funciones auxiliares, importar módulos o declarar constantes en python-code, pero la función eval que realiza la transformación de datos debe estar en el nivel superior del código. Por ejemplo:

import re

EMAIL_PATTERN = re.compile(r"\s+")

def normalize(value):
    return EMAIL_PATTERN.sub("", value).lower()

def eval(value: str) -> str:
    if value is None: return None
    return normalize(value)

Mapeo de tipos

Flink CDC determina el tipo de columna de salida de una UDF según la anotación de tipo de retorno de la función eval. Actualmente, solo se admiten los siguientes tipos de parámetros y valores de retorno:

Registrar una UDF de Python

Para registrar una UDF de Python, configure el parámetro python-code bajo pipeline.user-defined-function en un trabajo de ingesta de Flink CDC. La sintaxis de cadena multilínea | de YAML le permite escribir bloques de código Python con sangría de forma natural:

pipeline:
  user-defined-function:
    - name: normalize_email
      python-code: |
        def eval(value: str) -> str:
            if value is None:
                return None
            return value.strip().lower()

    - name: double_age
      python-code: |
        def eval(value: int) -> int:
            if value is None:
                return None
            return value * 2

En este caso, name es el nombre de la función que se utiliza en las expresiones de transformación y no necesita coincidir con el nombre de la función de Python. La función que Flink CDC invoca en el código en línea siempre es eval.

Manejo de valores nulos y excepciones en UDF de Python

Los valores NULL en un pipeline de ingesta corresponden a None en Python. Si se invoca una UDF con un argumento NULL, la función eval de Python recibe None. Si la función eval de Python devuelve None, se utiliza NULL como valor de retorno de la UDF.

Si la función eval de Python genera una excepción en tiempo de ejecución, el operador de transformación desencadena un failover. Si existe la posibilidad de que su UDF falle, capture las excepciones mediante instrucciones try/except de Python para mantener la estabilidad del trabajo.

Parámetros de UDF de Python

Las UDF de Python admiten los siguientes parámetros.

transform:
  - source-table: db\.*
    projection: "*, inc(inc(inc(id))) as inc_id, format(id, 'id -> %d') as formatted_id"
    filter: inc(id) < 100