当前位置：首页 > news >正文

Flink 系列之十四 - Data Stream API的自定义数据类型

news 来源：原创 2025/7/27 15:06:32

之前做过数据平台，对于实时数据采集，使用了Flink。现在想想，在数据开发平台中，Flink的身影几乎无处不在，由于之前是边用边学，总体有点混乱，借此空隙，整理一下Flink的内容，算是一个知识积累，同时也分享给大家。

注意：由于框架不同版本改造会有些使用的不同，因此本次系列中使用基本框架是 Flink-1.19.x，Flink支持多种语言，这里的所有代码都是使用java，JDK版本使用的是19。
代码参考：https://github.com/forever1986/flink-study.git

Flink中的不同算子可能在不同线程、进程甚至机子，因此它们之间的数据传输就需要进行序列化。Flink有一套自定义的数据类型以及序列化规则。在前面一直使用的是基本类型或者元组类型，并没有定义序列化类，这是因为这些Flink都已经帮用户实现了一些基本数据类型的序列化。今天这一章来好好讲一下Flink对数据类型的支持以及如何自定义数据类型。

1 Flink的TypeInformation

序列化的本质就是将数据结构或者对象转换成一个二进制串的过程，在 Java 里面可以简单地理解成一个 byte 数组。而反序列化恰恰相反，就是将序列化过程中所生成的二进制串转换成数据结构或者对象的过程。

在Flink中，它自定义了自己一套完整的序列化和子序列化过程。

首先，Flink会根据类型找到自己本身自定义的数据类型的序列化器，如果存在则直接使用
其次，用户可以通过registerType方法注册自己的数据类型序列化器，如果已经注册则使用
再者，用户可以通过自定义 TypeInfo 和 TypeInfoFactory来实现自己的数据类型序列化，如果实现了，则使用
以上都没有，则默认会交给 Kryo 处理（或者启用如Avro其它序列化框架）

1.1 TypeInformation的关系

前面所说的，Flink已经为用户实现了很多类型的序列化。其基础就是基于一个TypeInformation抽象类来实现不同数据类型的序列化，下面是根据TypeInformation源码的一些主要方法做一下解读：

public abstract class TypeInformation<T> implements Serializable {private static final long serialVersionUID = -7742311969684489493L;/*** 检查此类型信息是否表示基本类型。基本类型是Integer, Strings, Date, Void等等*/@PublicEvolvingpublic abstract boolean isBasicType();/*** 判断是否为元组类型*/@PublicEvolvingpublic abstract boolean isTupleType();/*** 获得没有嵌套的字段数*/@PublicEvolvingpublic abstract int getArity();/*** 获取该类型的字段数*/@PublicEvolvingpublic abstract int getTotalFields();/*** 返回该类型的Class*/@PublicEvolvingpublic abstract Class<T> getTypeClass();/*** 提供关于泛型类型参数到子类型的类型信息的映射，主要是泛化类型需要提供。*/@PublicEvolvingpublic Map<String, TypeInformation<?>> getGenericParameters() {return Collections.emptyMap();}/*** 该类型是否可以用作密钥。作为最低要求，类型必须是可哈希的，并且与键具有可比性。*/@PublicEvolvingpublic abstract boolean isKeyType();/*** 此类型是否可以用作排序的键。对这种类型进行排序产生的顺序必须是有意义的。*/@PublicEvolvingpublic boolean isSortKeyType() {return isKeyType();}/***   创建该类型的序列化器：TypeSerializer*/@PublicEvolvingpublic TypeSerializer<T> createSerializer(SerializerConfig config) {if (config != null) {ExecutionConfig executionConfig = ((SerializerConfigImpl) config).getExecutionConfig();return createSerializer(executionConfig);} else {return createSerializer((ExecutionConfig) null);}}/*** Flink提供typeHint方法为泛型指定类型，这个在讲typeHint使用时会说到*/public static <T> TypeInformation<T> of(TypeHint<T> typeHint) {return typeHint.getTypeInfo();}
}

1.2 Flink 内置的数据类型

Flink内置了很多数据类型，了解这些数据类型，就能够知道哪些情况下不需要自定义序列化：

在这里插入图片描述

根据上图，可以将Flink的内置数据类型分为几大类：

1）基础数据类型：主要处理基础数据类型，涵盖java的8大基础类型以及常见的void、String、Date、BigXXX、Instant，包括装箱或者未装箱，比如int、Integer等，其主要的实现类是BasicTypeInfo

2）数组类型：支持java的8大基础类型以及String。主要实现类是BasicArrayTypeInfo

3）元组类型：主要处理元组。Flink Tuples是固定长度固定类型的Java Tuple实现，不支持空值存储。主要实现类是TupleTypeInfo

4）辅助类型：主要处理列表、map等类型，其主要的实现类ListTypeInfo、MapTypeInfo、EitherTypeInfo等

5）复合类型：主要处理java中的元组、或者scala中的元组、Pojo等，其中Pojo需要符合一定规范。其主要的实现类TupleTypeInfo、RowTypeInfo、PojoTypeInfo等

6）泛型类型：主要是实现泛型类型，其主要的实现类有GenericTypeInfo。

7）其它类型：Flink还内置了很多其它的类型，这个使用到的再去看看。

知识点：上面Flink实现了很多基本类型的序列化工作，如果需要指定一定的TypeInformation数据类型，那么Flink提供了org.apache.flink.api.common.typeinfo.Types这个类，里面有很多基本TypeInformation数据类型常量，直接使用即可。

2 自定义类型的演示

在使用Flink过程中，与数据类型相关的无非就是以下三种场景，这里一一使用demo演示

代码参考：lesson07

1）新建子模块lesson07，其pom引入如下：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><scope>provided</scope>
</dependency>

2.1 returns()方法

虽然Flink已经为封装了很多类型，但是随着java的泛型的引入，基于历史原因以及性能改进就会出现泛型擦除的问题，在底层编译的时候，其实会对泛型进行擦除，导致JVM只能识别到是List，但是List中的数据是什么类型可能已经被擦除了，这个会导致运行时报错。

为了解决这个问题，Flink提供returns方法，在算子后面可以通过该方法，说明关于此操作符返回类型的类型信息提示。在Flink无法自动确定函数生成的类型的情况下，可以使用此方法。下面的案例来演示一下returns方法：

ReturnsDemo类：

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class ReturnsDemo {public static void main(String[] args) throws Exception {// 1. 创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2. 读取数据DataStreamSource<String> text = env.socketTextStream("127.0.0.1", 9999);// 3. 计算text.map((String value) -> {String[] values = value.split(",");String value1 = values[0];double value2 = Double.parseDouble("0");long value3 = 0;if (values.length >= 2) {try {value2 = Double.parseDouble(values[1]);} catch (Exception e) {value2 = Double.parseDouble("0");}}if (values.length >= 3) {try {value3 = Long.parseLong(values[2]);} catch (Exception ignored) {}}return new Tuple3<>(value1, value2, value3);})// 方法1：如果不知道返回类型，这里会报错.returns(Types.TUPLE(Types.STRING, Types.DOUBLE, Types.LONG))// 方法2：使用TypeHint方式
//                .returns(new TypeHint<Tuple3<String,Double,Long>> (){}).print();// 执行env.execute();}
}

当不加入returns() 方法这行代码时，会报以下错误
在这里插入图片描述

知识点：在java中，当返回的是泛型，同时使用Lambda表达式时，就会出现泛型擦除问题，这时候需要使用returns方法指定算子的返回类型。returns方法有2种传入类型方式：
1）直接使用Types已经定义好的TypeInformation方式
2）使用TypeHint生成对应的TypeInformation方式

2.2 Pojo类型

从前面了解到，Flink支持很多基础数据类型，对于自定义的类，也提供了PojoTypeInfo支持自定义类的序列化。但是使用PojoTypeInfo对自定义的类序列化，需要符合以下条件：

1）该类必须是public
2）该类的属性要么是public，要么提供标准的setter/getter方法
3）该类必须提供一个无参构造函数
4）如果该类的某个属性也是自定义的类，那么属性对应的这个类也要符合1,2,3,4规则

PojoTypeDemol类：

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class PojoTypeDemo {public static void main(String[] args) throws Exception {// 1. 创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2. 读取数据DataStreamSource<String> text = env.socketTextStream("127.0.0.1", 9999);// 3. 计算text.map((String value) -> {String[] values = value.split(",");String value1 = values[0];double value2 = Double.parseDouble("0");long value3 = 0;if (values.length >= 2) {try {value2 = Double.parseDouble(values[1]);} catch (Exception e) {value2 = Double.parseDouble("0");}}if (values.length >= 3) {try {value3 = Long.parseLong(values[2]);} catch (Exception ignored) {}}return new ServerInfo(value1, new CPUInfo(value2, "%"), value3);}).print();// 执行env.execute();}// 1.必须是publicpublic static class ServerInfo{// 2.属性要么是public，要么提供setter/getter方法public String serverId;// 3. 如果属性也是自定义的类型，那么这个类型也要符合1,2,3,4规则public CPUInfo cpuInfo;public Long time;// 4.必须提供一个无参构造函数public ServerInfo() {}public ServerInfo(String serverId, CPUInfo cpuInfo, Long time) {this.serverId = serverId;this.cpuInfo = cpuInfo;this.time = time;}public String getServerId() {return serverId;}public void setServerId(String serverId) {this.serverId = serverId;}public CPUInfo getCpuInfo() {return cpuInfo;}public void setCpuInfo(CPUInfo cpuInfo) {this.cpuInfo = cpuInfo;}public Long getTime() {return time;}public void setTime(Long time) {this.time = time;}}public static class CPUInfo{private Double cpu;private String unit;public CPUInfo() {}public CPUInfo(Double cpu, String unit) {this.cpu = cpu;this.unit = unit;}public Double getCpu() {return cpu;}public void setCpu(Double cpu) {this.cpu = cpu;}public String getUnit() {return unit;}public void setUnit(String unit) {this.unit = unit;}}
}

自定义的类如果不符合以上几点，则在运行时会报错：
在这里插入图片描述

2.3 自定义TypeInformation

Flink基本上能满足用户99.9%的需求，但如果Flink实现的序列化还不满足要求，用户可以自定义TypeInformation。

1）需要实现TypeInformation，并实现其方法
2）需要实现TypeSerializer，用于实现序列化

这里暂时没有想到一个实际场景，就不演示自定义TypeInformation

结语：本章通过讲解Flink的TypeInformation，对其数据类型也做了一个全面的了解。实际应用中或许涉及这方面的比较少，但是了解其使用原理，一旦遇到了就不会花费太多时间去找到问题所在。至此关于Data Stream API的章节就讲完了，接下来的章节，将了解Flink较为高级的概念。

目录