Write efficient user-defined functions-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

Flink's SQL engine operates with two data type systems at runtime: the Java types your user-defined function (UDF) code uses (such as Map and List) and the binary-optimized internal types the engine uses internally (such as MapData and ArrayData). By default, Flink converts between these systems on every UDF call, adding CPU and memory overhead. This topic explains how to write UDFs that operate directly on internal data types, eliminating that conversion cost.

Data type system

Type	Description
External data type	Java types exposed to user code: `Map`, `List`, `String`. These are familiar but require conversion before the engine can process them.
Internal data type	Binary representations optimized by the Flink engine: `MapData`, `ArrayData`, `RowData`. Operating on these directly avoids serialization and deserialization overhead.

Example: extract map keys

The following UDF extracts all keys from a MapData field and returns them as an ArrayData. Because the output element type depends on the input map's key type, this case requires getTypeInference().

Input and output

Input	Output
`SELECT mapkey(MAP['A',1,'B',2]);`	`[A, B]`
`SELECT mapkey(STR_TO_MAP('a=1,b=2'));`	`[a, b]`

Java

package com.aliyun.example;

import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.ConstantArgumentCount;
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.utils.DataTypeUtils;
import org.apache.flink.table.api.DataTypes;

import java.util.Optional;
import java.util.List;

public class MapKeyUDF extends ScalarFunction {

    // Accept MapData (not Java Map) and return ArrayData (not Java List).
    // No type conversion occurs on each call.
    public ArrayData eval(MapData input) {
        if (input == null) return null;
        return input.keyArray();
    }

    @Override
    public TypeInference getTypeInference(DataTypeFactory typeFactory) {
        return newBuilder()
            .inputTypeStrategy(MAP)
            .outputTypeStrategy(nullableIfArgs(MAP_KEYS))
            .build();
    }

    // Input type strategy: accept exactly one argument and convert it to an internal data type.
    private static final InputTypeStrategy MAP = new InputTypeStrategy() {
        @Override
        public ArgumentCount getArgumentCount() {
            return ConstantArgumentCount.of(1);
        }

        @Override
        public Optional<List<DataType>> inferInputTypes(CallContext callContext, boolean throwOnFailure) {
            return Optional.of(
                callContext.getArgumentDataTypes().stream()
                    .map(DataTypeUtils::toInternalDataType)
                    .collect(Collectors.toList()));
        }

        @Override
        public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
            return null;
        }
    };

    // Output type strategy: return an array whose element type matches the map's key type.
    private static final TypeStrategy MAP_KEYS = callContext ->
        Optional.of(
            DataTypeUtils.toInternalDataType(DataTypes.ARRAY(
                ((KeyValueDataType) callContext.getArgumentDataTypes().get(0)).getKeyDataType()
            ))
        );

    private static final TypeStrategy nullableIfArgs(TypeStrategy strategy) {
        return callContext -> strategy.infer(callContext).map(DataType::copy);
    }
}

Key implementation points

`eval` signature uses internal types. MapData replaces Map as the input, and ArrayData replaces List as the output. The engine passes data through without conversion.
Input type strategy converts arguments to internal types. DataTypeUtils::toInternalDataType ensures Flink treats each argument as an internal data type. ConstantArgumentCount.of(1) enforces the single-argument contract.
Output type strategy reflects the input's key type. Casting to KeyValueDataType and calling getKeyDataType() extracts the key type dynamically, so the function works with any map whose key type Flink can infer.

Maven dependencies

<!-- Flink Table Runtime -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-runtime</artifactId>
  <version>${flink.version}</version>
</dependency>

<!-- Flink Table Common -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-common</artifactId>
  <version>${flink.version}</version>
</dependency>

<!-- Flink Table API Java Bridge -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-java-bridge</artifactId>
  <version>${flink.version}</version>
</dependency>

Performance benefits and trade-offs

Benefits

No type conversion overhead. The engine passes data directly to your UDF without converting between Java and internal representations.
Lower garbage collection pressure. Fewer intermediate objects are created, reducing GC frequency.
Less memory usage. Internal binary representations are more compact than Java heap objects.
Improved data processing efficiency. Leverages Flink engine's specialized optimizations for internal data types.

Trade-offs

Narrower API surface. Internal types like MapData and ArrayData support fewer operations than their Java equivalents.
Steeper learning curve. You need to understand the Flink internal data type API before writing these UDFs.
Reduced readability and debuggability. Binary representations are harder to inspect than Java objects. Add code comments to compensate.

Best practices

Profile before optimizing. Use internal data types for UDFs that process large data volumes or run in tight loops. For low-throughput functions, the added complexity may not justify the gain.
Learn the core internal type APIs. Familiarize yourself with ArrayData, MapData, and RowData before writing production UDFs. The Flink JavaDoc and source code examples (flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/) are the most reliable references.
Implement type inference correctly. Override getTypeInference() and use DataTypeUtils::toInternalDataType to ensure both input and output are declared as internal types.
Avoid object creation in `eval`. Never instantiate new objects inside eval or any method called at high frequency. Use accessors like keyArray() that return views into the existing binary data.
Verify with performance tests. Compare your UDF against a Java-type equivalent under realistic data volumes to confirm the expected improvement.
Add comments. Internal type code is non-obvious. Document why each type is chosen and what each accessor returns.

What's next

上一篇: Exception handling guide 下一篇: Handle out-of-order changelog events