Flink SQL 知其所以然：SQL 数据类型大全！

SQL 数据类型

在介绍完一些基本概念之后，我们来认识一下，Flink SQL 中的数据类型。

Flink SQL 内置了很多常见的数据类型，并且也为用户提供了自定义数据类型的能力。

总共包含 3 部分：

原子数据类型。
复合数据类型。
用户自定义数据类型。

一、原子数据类型

1、字符串类型：

CHAR、CHAR(n)：定长字符串，就和 Java 中的 Char 一样，n 代表字符的定长，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。
VARCHAR、VARCHAR(n)、STRING：可变长字符串，就和 Java 中的 String 一样，n 代表字符的最大长度，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。STRING 等同于 VARCHAR(2147483647)。

2、二进制字符串类型：

BINARY、BINARY(n)：定长二进制字符串，n 代表定长，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。
VARBINARY、VARBINARY(n)、BYTES：可变长二进制字符串，n 代表字符的最大长度，取值范围 [1, 2,147,483,647]。如果不指定 n，则默认为 1。BYTES 等同于 VARBINARY(2147483647)。

3、精确数值类型：

DECIMAL、DECIMAL(p)、DECIMAL(p, s)、DEC、DEC(p)、DEC(p, s)、NUMERIC、NUMERIC(p)、NUMERIC(p, s)：固定长度和精度的数值类型，就和 Java 中的 BigDecima一样，p 代表数值位数(长度)，取值范围 [1, 38];s 代表小数点后的位数(精度)，取值范围 [0, p]。如果不指定，p 默认为 10，s 默认为 0。
TINYINT：-128 到 127 的 1 字节大小的有符号整数，就和 Java 中的 byte 一样。
SMALLINT：-32,768 to 32,767 的 2 字节大小的有符号整数，就和 Java 中的 short 一样。
INT、INTEGER：-2,147,483,648 to 2,147,483,647 的 4 字节大小的有符号整数，就和 Java 中的 int 一样。
BIGINT：-9,223,372,036,854,775,808 to 9,223,372,036,854,775,807 的 8 字节大小的有符号整数，就和 Java 中的 long 一样。

4、有损精度数值类型：

FLOAT：4 字节大小的单精度浮点数值，就和 Java 中的 float 一样。
DOUBLE、DOUBLE PRECISION：8 字节大小的双精度浮点数值，就和 Java 中的 double 一样。
关于 FLOAT 和 DOUBLE 的区别可见 https://www.runoob.com/w3cnote/float-and-double-different.html。

5、布尔类型：BOOLEAN。

6、NULL 类型：NULL。

7、Raw 类型：RAW(‘class’, ‘snapshot’) 。只会在数据发生网络传输时进行序列化，反序列化操作，可以保留其原始数据。以 Java 举例，class 参数代表具体对应的 Java 类型，snapshot 代表类型在发生网络传输时的序列化器。

8、日期、时间类型：

DATE：由年-月-日组成的不带时区含义的日期类型，取值范围 [0000-01-01, 9999-12-31]
TIME、TIME(p)：由小时：分钟：秒[.小数秒] 组成的不带时区含义的的时间的数据类型，精度高达纳秒，取值范围 [00:00:00.000000000到23:59:59.9999999]。其中 p 代表小数秒的位数，取值范围 [0, 9]，如果不指定 p，默认为 0。
TIMESTAMP、TIMESTAMP(p)、TIMESTAMP WITHOUT TIME ZONE、TIMESTAMP(p) 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(p) 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(p)：由年-月-日小时：分钟：秒[.小数秒] 时区组成的带时区含义的时间类型，取值范围 [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 WITH TIME ZONE 的区别在于：TIMESTAMP WITH TIME ZONE 的时区信息是携带在数据中的，举例：其输入数据应该是 2022-01-01 00:00:00.000000000 +08:00;TIMESTAMP_LTZ 的时区信息不是携带在数据中的，而是由 Flink SQL 任务的全局配置决定的，我们可以由 table.local-time-zone 参数来设置时区。
INTERVAL YEAR TO MONTH、 INTERVAL DAY TO SECOND：interval 的涉及到的种类比较多。INTERVAL 主要是用于给 TIMESTAMP、TIMESTAMP_LTZ 添加偏移量的。举例，比如给 TIMESTAMP 加、减几天、几个月、几年。INTERVAL 子句总共涉及到的语法种类如下 Flink SQL 案例所示。

CREATETABLEsink_table (
result_interval_yearTIMESTAMP(3),
result_interval_year_pTIMESTAMP(3),
result_interval_year_p_to_monthTIMESTAMP(3),
result_interval_monthTIMESTAMP(3),
result_interval_dayTIMESTAMP(3),
result_interval_day_p1TIMESTAMP(3),
result_interval_day_p1_to_hourTIMESTAMP(3),
result_interval_day_p1_to_minuteTIMESTAMP(3),
result_interval_day_p1_to_second_p2TIMESTAMP(3),
result_interval_hourTIMESTAMP(3),
result_interval_hour_to_minuteTIMESTAMP(3),
result_interval_hour_to_secondTIMESTAMP(3),
result_interval_minuteTIMESTAMP(3),
result_interval_minute_to_second_p2TIMESTAMP(3),
result_interval_secondTIMESTAMP(3),
result_interval_second_p2TIMESTAMP(3)
) WITH (
'connector' = 'print'
);
INSERTINTOsink_table
SELECT
--FlinkSQL 支持的所有 INTERVAL 子句如下，总体可以分为 `年-月`、`日-小时-秒` 两种

--1. 年-月。取值范围为 [-9999-11, +9999-11]，其中 p 是指有效位数，取值范围 [1, 4]，默认值为 2。比如如果值为 1000，但是 p = 2，则会直接报错。
--INTERVALYEAR
f1 + INTERVAL'10'YEARasresult_interval_year
--INTERVALYEAR(p)
    , f1 + INTERVAL'100'YEAR(3) asresult_interval_year_p
--INTERVALYEAR(p) TOMONTH
    , f1 + INTERVAL'10-03'YEAR(3) TOMONTHasresult_interval_year_p_to_month
--INTERVALMONTH
    , f1 + INTERVAL'13'MONTHasresult_interval_month

--2. 日-小时-秒。取值范围为 [-99999923:59:59.999999999, +99999923:59:59.999999999]，其中 p1\p2 都是有效位数，p1 取值范围 [1, 6]，默认值为 2；p2 取值范围 [0, 9]，默认值为 6
--INTERVALDAY
    , f1 + INTERVAL'10'DAYasresult_interval_day
--INTERVALDAY(p1)
    , f1 + INTERVAL'100'DAY(3) asresult_interval_day_p1
--INTERVALDAY(p1) TOHOUR
    , f1 + INTERVAL'10 03'DAY(3) TOHOURasresult_interval_day_p1_to_hour
--INTERVALDAY(p1) TOMINUTE
    , f1 + INTERVAL'10 03:12'DAY(3) TOMINUTEasresult_interval_day_p1_to_minute
--INTERVALDAY(p1) TOSECOND(p2)
    , f1 + INTERVAL'10 00:00:00.004'DAYTOSECOND(3) asresult_interval_day_p1_to_second_p2
--INTERVALHOUR
    , f1 + INTERVAL'10'HOURasresult_interval_hour
--INTERVALHOURTOMINUTE
    , f1 + INTERVAL'10:03'HOURTOMINUTEasresult_interval_hour_to_minute
--INTERVALHOURTOSECOND(p2)
    , f1 + INTERVAL'00:00:00.004'HOURTOSECOND(3) asresult_interval_hour_to_second
--INTERVALMINUTE
    , f1 + INTERVAL'10'MINUTEasresult_interval_minute
--INTERVALMINUTETOSECOND(p2)
    , f1 + INTERVAL'05:05.006'MINUTETOSECOND(3) asresult_interval_minute_to_second_p2
--INTERVALSECOND
    , f1 + INTERVAL'3'SECONDasresult_interval_second
--INTERVALSECOND(p2)
    , f1 + INTERVAL'300'SECOND(3) asresult_interval_second_p2
FROM (SELECTTO_TIMESTAMP_LTZ(1640966476500, 3) asf1)

二、复合数据类型

数组类型：ARRAY、t ARRAY。数组最大长度为 2,147,483,647。t 代表数组内的数据类型。举例 ARRAY、ARRAY，其等同于 INT ARRAY、STRING ARRAY。
Map 类型：MAP。Map 类型就和 Java 中的 Map 类型一样，key 是没有重复的。举例 Map、Map。
集合类型：MULTISET、t MULTISET。就和 Java 中的 List 类型，一样，运行重复的数据。举例 MULTISET，其等同于 INT MULTISET。
对象类型：ROW、ROW、ROW(n0 t0, n1 t1, …>、ROW(n0 t0 ‘d0’, n1 t1 ‘d1’, …)。就和 Java 中的自定义对象一样。举例：ROW(myField INT, myOtherField BOOLEAN)，其等同于 ROW。

三、用户自定义数据类型

用户自定义类型就是运行用户使用 Java 等语言自定义一个数据类型出来。但是目前数据类型不支持使用 CREATE TABLE 的 DDL 进行定义，只支持作为函数的输入输出参数。如下案例：

第一步，自定义数据类型

publicclassUser {
    // 1. 基础类型，Flink 可以通过反射类型信息自动把数据类型获取到
    // 关于 SQL 类型和 Java 类型之间的映射见：https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/types/#data-type-extraction
publicintage;
publicStringname;
    // 2. 复杂类型，用户可以通过 @DataTypeHint("DECIMAL(10, 2)") 注解标注此字段的数据类型
public@DataTypeHint("DECIMAL(10, 2)") BigDecimaltotalBalance;
}

第二步，在 UDF 中使用此数据类型

publicclassUserScalarFunctionextendsScalarFunction {
    // 1. 自定义数据类型作为输出参数
publicUsereval(longi) {
if (i > 0 && i <= 5) {
Useru = newUser();
u.age = (int) i;
u.name = "name1";
u.totalBalance = newBigDecimal(1.1d);
returnu;
        } else {
Useru = newUser();
u.age = (int) i;
u.name = "name2";
u.totalBalance = newBigDecimal(2.2d);
returnu;
        }
    }    
    // 2. 自定义数据类型作为输入参数
publicStringeval(Useri) {
if (i.age > 0 && i.age <= 5) {
Useru = newUser();
u.age = 1;
u.name = "name1";
u.totalBalance = newBigDecimal(1.1d);
returnu.name;
        } else {
Useru = newUser();
u.age = 2;
u.name = "name2";
u.totalBalance = newBigDecimal(2.2d);
returnu.name;
        }
    }
}

第三步，在 Flink SQL 中使用

--1. 创建 UDF
CREATEFUNCTIONuser_scalar_funcAS'flink.examples.sql._12_data_type._02_user_defined.UserScalarFunction';
--2. 创建数据源表
CREATETABLEsource_table (
user_idBIGINTNOTNULLCOMMENT'用户 id'
) WITH (
'connector' = 'datagen',
'rows-per-second' = '1',
'fields.user_id.min' = '1',
'fields.user_id.max' = '10'
);

--3. 创建数据汇表
CREATETABLEsink_table (
result_row_1ROW<ageINT, nameSTRING, totalBalanceDECIMAL(10, 2)>,
result_row_2STRING
) WITH (
'connector' = 'print'
);
--4.SQL 查询语句
INSERTINTOsink_table
select
--4.a. 用户自定义类型作为输出
user_scalar_func(user_id) asresult_row_1,
--4.b. 用户自定义类型作为输出及输入
user_scalar_func(user_scalar_func(user_id)) asresult_row_2
fromsource_table;
--5. 查询结果
+I[+I[9, name2, 2.20], name2]
+I[+I[1, name1, 1.10], name1]
+I[+I[5, name1, 1.10], name1]