Flink SQL无法将RAW('org.apache.avro.util.Utf8')转换为String问题咨询

问题根因 #

Flink从1.13版本开始调整了Avro格式的默认类型映射规则：1.12及更早版本会自动将Avro内置的org.apache.avro.util.Utf8类型直接映射为Flink Table的STRING类型；1.13+版本默认关闭了这个自动转换，未做特殊配置时会将Utf8类识别为未知的自定义类型，封装为RAW类型存储，导致后续类型转换失败。

解决方案 #

方案1：修改Source表Avro配置开启自动转换（最推荐） #

在定义读取Kafka数据的Source表的WITH参数中，新增配置项'avro.convert.utf8.to.string' = 'true'，开启Utf8到STRING的自动映射，无需修改后续业务逻辑。
示例Source表配置补充：

CREATE TABLE customer (
  `customerId` STRING,
  `age` INT
) WITH (
  'connector' = 'kafka',
  'topic' = '你的源topic名',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'value.format' = 'avro',
  -- 新增以下配置
  'avro.convert.utf8.to.string' = 'true'
)

配置生效后读取到的customerId字段会直接是STRING类型，原有INSERT逻辑无需修改即可正常运行。

方案2：自定义UDF实现RAW类型转STRING #

如果无法修改Source表配置，可以实现一个简单的UDF，调用Utf8类的toString()方法完成转换，避开原生CAST函数不支持RAW类型转STRING的限制。

1. 定义UDF类：

import org.apache.avro.util.Utf8;
import org.apache.flink.table.functions.ScalarFunction;

public class Utf8ToString extends ScalarFunction {
    public String eval(Utf8 utf8) {
        return utf8 == null ? null : utf8.toString();
    }
}

2. 注册UDF后调用：

// 注册UDF
tableEnv.createTemporarySystemFunction("Utf8ToString", Utf8ToString.class);
// 改写INSERT语句
TableResult table3 = tableEnv.executeSql(
    "insert into SIMPLECUSTOMER select Utf8ToString(customerId), avg(age) as age from customer group by customerId "
);

方案3：调整Avro生成类的配置 #

如果是用Avro Maven插件自动生成的Java类，可以在插件配置中添加<stringType>String</stringType>参数，让生成的类中字符串字段直接用java.lang.String类型，而非Utf8类型，从根源避免类型问题。
插件配置示例：

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>${avro.version}</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
                <outputDirectory>${project.build.directory}/generated-sources/avro/</outputDirectory>
                <!-- 新增以下配置 -->
                <stringType>String</stringType>
            </configuration>
        </execution>
    </executions>
</plugin>

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