Flink CDC UDFs

Type mappings

The following table maps CDC column types to the corresponding Java types used in eval method parameters, return values, and getReturnType:

The VARIANT class path differs from the Variant class path in Flink SQL.

Syntax rules

A Python UDF must meet the following requirements:

• A top-level function named eval must be defined in python-code.

• The eval function must declare a return type annotation.

• The number of parameters of the eval function must match the number of arguments passed when the UDF is called.

• Parameter type annotations are not required, but keeping them is recommended for code readability.

• Python UDFs currently support only scalar functions. Each invocation returns a single value.

• SQL NULL values in the input data are converted to Python None.

• A Python function can return None. In that case, the output is SQL NULL.

For example, the following Python UDF trims whitespace from the input string and converts it to lowercase:

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

You can define helper functions, import modules, or declare constants in python-code, but the eval function that performs the data transformation must be at the top level of the code. For example:

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)

Type mappings

Flink CDC determines the output column type of a UDF based on the return type annotation of the eval function. Currently, only the following parameter types and return value types are supported:

Register a Python UDF

To register a Python UDF, configure the python-code parameter under pipeline.user-defined-function of a Flink CDC ingestion job. The YAML | multi-line string syntax lets you write indented Python code blocks naturally:

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

Here, name is the function name used in transform expressions and does not need to match the Python function name. The function that Flink CDC calls in the inline code is always eval.

Null and exception handling of Python UDFs

NULL values in an ingestion pipeline correspond to Python None. If a UDF is called with a NULL argument, the Python eval function receives None. If the Python eval function returns None, NULL is used as the UDF return value.

If the Python eval function throws a runtime exception, the transform operator triggers a failover. If your UDF may fail, catch exceptions with Python try/except statements to keep the job stable.

Python UDF parameters

Python UDFs support the following parameters.