Apache Flink——Flink 支持的数据类型

一、Flink 的类型系统

Flink 作为一个分布式处理框架，处理的是以数据对象作为元素的流。如果用水流来类比，那么我们要处理的数据元素就是随着水流漂动的物体。在这条流动的河里，可能漂浮着小木块，也可能行驶着内部错综复杂的大船。要分布式地处理这些数据，就不可避免地要面对数据的网络传输、状态的落盘和故障恢复等问题，这就需要对数据进行序列化和反序列化。小木块是容易序列化的；而大船想要序列化之后传输，就需要将它拆解、清晰地知道其中每一个零件的类型。

为了方便地处理数据，Flink 有自己一整套类型系统。Flink 使用“类型信息”（TypeInformation）来统一表示数据类型。TypeInformation 类是 Flink 中所有类型描述符的基类。它涵盖了类型的一些基本属性，并为每个数据类型生成特定的序列化器、反序列化器和比较器。

二、Flink 支持的数据类型

简单来说，对于常见的 Java 和 Scala 数据类型，Flink 都是支持的。Flink 在内部，Flink
对支持不同的类型进行了划分，这些类型可以在 Types 工具类中找到：

• 1、基本类型
  所有 Java 基本类型及其包装类，再加上 Void、String、Date、BigDecimal 和 BigInteger

• 2、数组类型
  包括基本类型数组（PRIMITIVE_ARRAY）和对象数组(OBJECT_ARRAY)

• 3、复合数据类型

• Java 元组类型（TUPLE）：这是 Flink 内置的元组类型，是 Java API 的一部分。最多
  25 个字段，也就是从 Tuple0~Tuple25，不支持空字段。

• Scala 样例类及 Scala 元组：不支持空字段。

• 行类型（ROW）：可以认为是具有任意个字段的元组,并支持空字段。

• POJO：Flink 自定义的类似于 Java bean 模式的类。

• 4、辅助类型
  Option、Either、List、Map 等

• 5、泛型类型（GENERIC）

Flink 支持所有的 Java 类和 Scala 类。不过如果没有按照上面 POJO 类型的要求来定义，就会被 Flink 当作泛型类来处理。Flink 会把泛型类型当作黑盒，无法获取它们内部的属性；它们也不是由 Flink 本身序列化的，而是由 Kryo 序列化的。

在这些类型中，元组类型和 POJO 类型最为灵活，因为它们支持创建复杂类型。而相比之下，POJO 还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。所以，在项目实践中，往往会将流处理程序中的元素类型定为 Flink 的 POJO 类型。

Flink 对 POJO 类型的要求如下：

• 类是公共的（public）和独立的（standalone，也就是说没有非静态的内部类）；
• 类有一个公共的无参构造方法；
• 类中的所有字段是 public 且非 final 的；或者有一个公共的 getter 和 setter 方法，这些
  方法需要符合 Java bean 的命名规范。

三、类型提示（Type Hints）

Flink 还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应的序列化器和反序列化器。但是，由于 Java 中泛型擦除的存在，在某些特殊情况下（比如 Lambda 表达式中），自动提取的信息是不够精细的——只告诉 Flink 当前的元素由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。为了解决这类问题，Java API 提供了专门的“类型提示”（type hints）。

回忆一下之前的 word count 流处理程序，我们在将 String 类型的每个词转换成（word，count）二元组后，就明确地用 returns 指定了返回的类型。因为对于 map 里传入的 Lambda 表达式，系统只能推断出返回的是 Tuple2 类型，而无法得到 Tuple2<String, Long>。只有显式地告诉系统当前的返回类型，才能正确地解析出完整数据。

.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));

这是一种比较简单的场景，二元组的两个元素都是基本数据类型。那如果元组中的一个元素又有泛型，该怎么处理呢？

Flink 专门提供了 TypeHint 类，它可以捕获泛型的类型信息，并且一直记录下来，为运行时提供足够的信息。我们同样可以通过.returns()方法，明确地指定转换之后的 DataStream 里元素的类型。

returns(new TypeHint<Tuple2<Integer, SomeType>>(){})

参考：
https://blog.csdn.net/weixin_45417821/article/details/124152670