如何基于Avro Schema创建Flink SQL UDF解码Kafka的load字段

一、前置依赖准备 #

确保你的Flink项目中引入以下Maven依赖（Gradle项目对应调整依赖配置）：

<dependencies>
    <!-- Flink Avro 核心依赖 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-avro</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <!-- Avro 基础依赖 -->
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.11.0</version>
    </dependency>
</dependencies>

二、两种UDF实现方式 #

方式1：基于生成的Avro实体类解码 #

如果已经通过Avro工具从data.avsc生成了对应Java实体类（例如命名为DataRecord），可以直接用实体类做解码载体：

1. 编写UDF类

import org.apache.flink.table.functions.ScalarFunction;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DecoderFactory;
import java.io.IOException;

public class AvroDecoderUDF extends ScalarFunction {
    private final GenericDatumReader<DataRecord> datumReader;

    public AvroDecoderUDF() {
        // 直接使用生成类内置的Schema
        this.datumReader = new GenericDatumReader<>(DataRecord.SCHEMA$);
    }

    public DataRecord eval(byte[] loadBytes) throws IOException {
        if (loadBytes == null || loadBytes.length == 0) {
            return null;
        }
        BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(loadBytes, null);
        return datumReader.read(null, decoder);
    }
}

2. 生成Avro实体类的Maven插件配置
   若还未生成实体类，在项目pom.xml中添加插件自动生成：

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-maven-plugin</artifactId>
            <version>1.11.0</version>
            <executions>
                <execution>
                    <phase>generate-sources</phase>
                    <goals>
                        <goal>schema</goal>
                    </goals>
                    <configuration>
                        <sourceDirectory>${project.basedir}/src/main/resources/</sourceDirectory>
                        <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

执行mvn generate-sources命令即可从src/main/resources/data.avsc生成实体类。

方式2：基于GenericRecord通用解码（无需生成实体类） #

如果不想生成实体类，可直接用GenericRecord处理任意Avro Schema：

import org.apache.flink.table.functions.ScalarFunction;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DecoderFactory;
import java.io.IOException;
import java.io.InputStream;

public class GenericAvroDecoderUDF extends ScalarFunction {
    private final GenericDatumReader<GenericRecord> datumReader;

    public GenericAvroDecoderUDF(String schemaPath) throws IOException {
        // 从资源文件加载data.avsc
        InputStream schemaStream = getClass().getClassLoader().getResourceAsStream(schemaPath);
        Schema schema = new Schema.Parser().parse(schemaStream);
        this.datumReader = new GenericDatumReader<>(schema);
    }

    public GenericRecord eval(byte[] loadBytes) throws IOException {
        if (loadBytes == null || loadBytes.length == 0) {
            return null;
        }
        BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(loadBytes, null);
        return datumReader.read(null, decoder);
    }
}

三、注册UDF并在Flink SQL中使用 #

1. 代码中注册并使用 #

在Flink Table API+SQL代码中注册UDF后直接查询：

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class AvroDecoderDemo {
    public static void main(String[] args) throws Exception {
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 注册方式1的UDF
        tableEnv.createTemporarySystemFunction("decode_avro", AvroDecoderUDF.class);
        // 或者注册方式2的UDF（需传入resources下的schema文件路径）
        // tableEnv.createTemporarySystemFunction("decode_avro", new GenericAvroDecoderUDF("data.avsc"));

        // 执行解码查询
        tableEnv.executeSql("SELECT `time`, metadata, decode_avro(load) AS decoded_load FROM dataSQL").print();
    }
}

2. Flink SQL客户端中注册使用 #

将UDF打包成jar包后，在SQL客户端中注册调用：

-- 注册UDF
CREATE FUNCTION decode_avro AS 'com.your.package.AvroDecoderUDF' USING JAR 'file:///path/to/your-udf.jar';

-- 查询解码后的数据
SELECT `time`, metadata, decode_avro(load) AS decoded_load FROM dataSQL;

四、注意事项 #

• 确保data.avsc的Schema与Kafka中实际存储的Avro数据Schema完全一致，否则会解码失败。
• 若使用Schema Registry管理Avro Schema，需在UDF中加入Schema Registry的客户端逻辑（如CachedSchemaRegistryClient）获取对应版本的Schema，当前场景基于本地Schema文件无需此步骤。
• 需处理空值、异常情况，避免UDF抛出未捕获异常导致任务崩溃。

内容的提问来源于stack exchange，提问作者Hsgh775