Flink 1.15中java.util.HashMap与java.time.Duration非有效POJO类型咨询

你遇到的日志提示是因为Flink无法将java.time.Duration和java.util.HashMap自动识别为合法POJO类型，会退化为GenericType使用Kryo序列化，这会影响性能。不需要替换成其他类，通过显式声明类型信息就能解决问题：

针对java.time.Duration的处理 #

Flink 1.15原生支持java.time.Duration，但需要显式指定类型信息让Flink识别为POJO：

1. 字段注解方式：在POJO的Duration字段上添加类型注解

import org.apache.flink.api.common.typeinfo.TypeInfo;
import org.apache.flink.api.java.typeutils.runtime.java.time.JavaTimeTypeInfoFactory;
import java.time.Duration;

public class YourPojo {
    @TypeInfo(JavaTimeTypeInfoFactory.DurationTypeInfoFactory.class)
    private Duration processTime;

    // 必须提供公共无参构造方法
    public YourPojo() {}

    // 提供字段的getter和setter
    public Duration getProcessTime() {
        return processTime;
    }

    public void setProcessTime(Duration processTime) {
        this.processTime = processTime;
    }
}

2. DataStream类型声明：在定义数据流时显式指定返回类型

DataStream<YourPojo> dataStream = env.fromCollection(yourDataList)
    .returns(TypeInformation.of(YourPojo.class));

针对java.util.HashMap的处理 #

java.util.HashMap本身符合Flink的POJO要求，但如果是泛型HashMap且未指定具体类型参数，Flink会无法识别。解决方式同样是显式声明泛型类型：

1. 字段注解方式：指定HashMap的泛型类型

import org.apache.flink.api.common.typeinfo.TypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfoFactory;
import java.util.HashMap;

public class YourPojo {
    @TypeInfo(MapTypeInfoFactory.class)
    private HashMap<String, Integer> metrics;

    public YourPojo() {}

    public HashMap<String, Integer> getMetrics() {
        return metrics;
    }

    public void setMetrics(HashMap<String, Integer> metrics) {
        this.metrics = metrics;
    }
}

2. 使用Types工具类声明类型：如果是直接处理HashMap数据流

import org.apache.flink.api.common.typeinfo.Types;
import java.util.HashMap;

DataStream<HashMap<String, Integer>> mapStream = env.fromCollection(mapDataList)
    .returns(Types.MAP(Types.STRING, Types.INT));

替代方案（不推荐） #

如果一定要用替代类：

• 对于Duration：可以用long类型存储毫秒/纳秒数，但会丢失Duration的语义，需要手动转换。
• 对于HashMap：可以使用org.apache.flink.api.java.tuple.Tuple类来存储键值对，但灵活性远不如HashMap，只适合固定数量的键值场景。

注意：显式声明类型信息是最优解，既保留原生类的语义，又能让Flink使用高效的POJO序列化器，避免性能损耗。

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