FLINK SQL数据类型

Flink SQL支持非常完善的数据类型，以满足不同的数据处理需求。以下是对Flink SQL数据类型的详细归纳：

一、原子数据类型

字符串类型
- CHAR、CHAR(n)：定长字符串，n代表字符的定长，取值范围为[1, 2147483647]。如果不指定n，则默认为1。
- VARCHAR、VARCHAR(n)、STRING：可变长字符串，n代表字符的最大长度，取值范围为[1, 2147483647]。如果不指定n，则默认为1。STRING等同于VARCHAR(2147483647)。
二进制字符串类型
- BINARY、BINARY(n)：定长二进制字符串，n代表定长，取值范围为[1, 2147483647]。如果不指定n，则默认为1。
- VARBINARY、VARBINARY(n)、BYTES：可变长二进制字符串，n代表字符的最大长度，取值范围为[1, 2147483647]。如果不指定n，则默认为1。BYTES等同于VARBINARY(2147483647)。
精确数值类型
- DECIMAL、DECIMAL§、DECIMAL(p, s)、DEC、DEC§、DEC(p, s)、NUMERIC、NUMERIC§、NUMERIC(p, s)：固定长度和精度的数值类型，p代表数值位数（长度），取值范围为[1, 38]；s代表小数点后的位数（精度），取值范围为[0, p]。如果不指定，p默认为10，s默认为0。
有损精度数值类型
- TINYINT：-128到127的1字节大小的有符号整数。
- SMALLINT：-32768到32767的2字节大小的有符号整数。
- INT、INTEGER：-2147483648到2147483647的4字节大小的有符号整数。
- BIGINT：-9223372036854775808到9223372036854775807的8字节大小的有符号整数。
浮点类型
- FLOAT：4字节大小的单精度浮点数值。
- DOUBLE、DOUBLE PRECISION：8字节大小的双精度浮点数值。
布尔类型
- BOOLEAN。
日期、时间类型
- DATE：由年-月-日组成的不带时区含义的日期类型，取值范围为[0000-01-01, 9999-12-31]。
- TIME、TIME§：由小时:分钟:秒[.小数秒]组成的不带时区含义的时间数据类型，精度高达纳秒，取值范围为[00:00:00.000000000, 23:59:59.9999999]。其中p代表小数秒的位数，取值范围为[0, 9]，如果不指定p，默认为0。
- TIMESTAMP、TIMESTAMP§、TIMESTAMP WITHOUT TIME ZONE、TIMESTAMP§ WITHOUT TIME ZONE：由年-月-日小时:分钟:秒[.小数秒]组成的不带时区含义的时间类型，取值范围为[0000-01-01 00:00:00.000000000, 9999-12-31 23:59:59.999999999]。其中p代表小数秒的位数，取值范围为[0, 9]，如果不指定p，默认为6。
- TIMESTAMP WITH TIME ZONE、TIMESTAMP§ WITH TIME ZONE：由年-月-日小时:分钟:秒[.小数秒] 时区组成的带时区含义的时间类型，取值范围为[0000-01-01 00:00:00.000000000 +14:59, 9999-12-31 23:59:59.999999999 -14:59]。其中p代表小数秒的位数，取值范围为[0, 9]，如果不指定p，默认为6。
- TIMESTAMP_LTZ、TIMESTAMP_LTZ§：与TIMESTAMP WITH TIME ZONE类似，但时区信息不是携带在数据中的，而是由Flink SQL任务的全局配置决定的。

二、复合数据类型

ARRAY：数组类型，类似于Java的array。
MULTISET：集合类型，类似于Java的List。
ROW：对象类型，可以包含多个字段，每个字段有自己的类型和名称，类似于Java的Object或Scala的Case Class。
MAP：映射类型，包含键值对，键和值都可以是任意类型。
样例：

Tuple（元组）

// 创建一个包含String和Integer类型字段的Tuple2  
DataStream<Tuple2<String, Integer>> tupleStream = env.fromElements(  new Tuple2<>("Alice", 30),  new Tuple2<>("Bob", 25)  
);  // 访问Tuple2的字段  
tupleStream.map(tuple -> tuple.f0 + " is " + tuple.f1 + " years old")  .print();

POJO（Plain Old Java Object，普通旧Java对象）

// 定义一个POJO类  
public class PersonPOJO {  public String name;  public int age;  // 无参构造方法  public PersonPOJO() {}  // 有参构造方法  public PersonPOJO(String name, int age) {  this.name = name;  this.age = age;  }  // Getter和Setter方法  public String getName() {  return name;  }  public void setName(String name) {  this.name = name;  }  public int getAge() {  return age;  }  public void setAge(int age) {  this.age = age;  }  
}  // 创建一个包含PersonPOJO对象的DataStream  
DataStream<PersonPOJO> personPOJOStream = env.fromElements(  new PersonPOJO("Alice", 30),  new PersonPOJO("Bob", 25)  
);  // 对DataStream进行处理  
personPOJOStream.map(person -> person.getName() + " is " + person.getAge() + " years old")  .print();

CREATE TABLE person_table (  id BIGINT,  name STRING,  age INT  
) WITH (  'connector' = '...,  ...  
);  -- 插入数据（假设已经有数据插入到person_table中）  -- 查询数据，并使用Row来表示结果集中的行  
SELECT id, name, age FROM person_table AS row(id BIGINT, name STRING, age INT);

Map 和 Array

// 创建一个包含Map和Array的DataStream  
DataStream<Tuple2<Map<String, String>, Integer[]>> complexStream = env.fromElements(  new Tuple2<>(  new HashMap<String, String>() {{  put("key1", "value1");  put("key2", "value2");  }},  new Integer[]{1, 2, 3}  ),  // ... 其他元素  
);  // 对DataStream进行处理  
complexStream.map(tuple -> {  Map<String, String> map = tuple.f0;  Integer[] array = tuple.f1;  // ... 对map和array进行处理  return "Processed result"; // 示例返回值，实际应根据需求返回合适的类型  
})  
.print();

三、用户自定义数据类型

Flink SQL也支持用户自定义数据类型，用户可以根据自己的需求定义复杂的数据结构，并通过实现相应的接口或类来注册这些自定义类型。
1、定义与用途
用户自定义数据类型通常用于处理那些无法直接通过Flink内置数据类型表示的数据。例如，当需要处理一个包含多个字段的复杂数据结构时，就可以定义一个包含这些字段的用户自定义数据类型。

2、实现方式
在Flink中，实现用户自定义数据类型通常需要遵循以下步骤：

定义数据类型：首先，需要定义一个Java或Scala类来表示用户自定义数据类型。这个类应该包含所有需要的字段，并提供相应的getter和setter方法（如果是Java类，还需要一个无参构造方法）。
实现序列化与反序列化：为了使Flink能够处理用户自定义数据类型，需要实现相应的序列化器和反序列化器。这些序列化器和反序列化器负责将用户自定义数据类型转换为字节流，以及从字节流中恢复出用户自定义数据类型。
注册类型信息：在Flink中注册用户自定义数据类型的类型信息。这通常是通过在Flink的配置中指定类型信息的方式来实现的。
3、注意事项
性能考虑：自定义数据类型的序列化与反序列化过程可能会对性能产生影响。因此，在实现自定义数据类型时，需要仔细考虑如何优化序列化与反序列化过程，以提高性能。
兼容性：当在不同的Flink集群或版本之间迁移时，需要确保自定义数据类型及其序列化器与反序列化器是兼容的。否则，可能会导致数据无法正确解析或处理。
错误处理：在处理用户自定义数据类型时，需要特别注意错误处理。例如，当遇到无法解析的数据时，应该能够优雅地处理这些错误，而不是导致整个作业失败。
4、应用场景
用户自定义数据类型在Flink中有广泛的应用场景。例如：
复杂数据结构处理：当需要处理包含多个字段的复杂数据结构时，可以使用用户自定义数据类型来表示这些结构。
自定义聚合函数：在实现自定义聚合函数时，可能需要使用用户自定义数据类型来存储中间结果或最终结果。
与外部系统交互：当Flink与外部系统（如数据库、消息队列等）交互时，可能需要将这些系统的数据类型转换为Flink能够处理的数据类型。这时，可以使用用户自定义数据类型来实现这种转换。
5、样例
1. 定义自定义数据类型
  首先，定义一个Java类来表示自定义数据类型。例如，我们定义一个名为Person的类，包含name和age两个字段。

public class Person {  private String name;  private int age;  // 无参构造方法  public Person() {}  // 有参构造方法  public Person(String name, int age) {  this.name = name;  this.age = age;  }  // Getter和Setter方法  public String getName() {  return name;  }  public void setName(String name) {  this.name = name;  }  public int getAge() {  return age;  }  public void setAge(int age) {  this.age = age;  }  // 重写toString方法，方便打印输出  @Override  public String toString() {  return "Person{name='" + name + "', age=" + age + "}";  }  
}

2. 实现序列化与反序列化

为了使Flink能够处理Person类型的数据，需要实现相应的序列化器和反序列化器。在Flink中，这通常通过实现TypeSerializer和TypeDeserializer接口来完成。然而，对于简单的POJO（Plain Old Java Object）类型，Flink通常能够自动推断并处理其序列化与反序列化过程，因此在这个例子中我们不需要显式实现这些接口。
3. 注册类型信息（如果需要）
在某些情况下，可能需要显式地在Flink中注册自定义数据类型的类型信息。这通常是在使用低级别的API（如DataStream API）时需要的。然而，在使用Table API或SQL时，Flink通常能够自动推断数据类型，因此不需要显式注册。
4. 使用自定义数据类型
现在，我们可以在Flink作业中使用Person类型的数据了。例如，我们可以创建一个DataStream，并向其中添加Person对象，然后对其进行处理。

// 假设已经有一个执行环境executionEnvironment  
DataStream<Person> personStream = executionEnvironment  .fromElements(new Person("Alice", 30), new Person("Bob", 25))  .name("Person Stream");  // 对DataStream进行处理，例如打印输出  
personStream.print();