Flink SQL中的数据类型

Flink SQL 为用户提供了一系列丰富的原始数据类型。

数据类型

在 Flink 的 Table 生态系统中，数据类型 描述了数据的逻辑类型，可以用来表示转换过程中输入、输出的类型。

Flink 的数据类型类似于 SQL 标准中的术语数据类型，但包含了值的可空性，以便于更好地处理标量表达式。

以下是一些数据类型的例子：

• INT
• INT NOT NULL
• INTERVAL DAY TO SECOND(3)
• ROW<myField ARRAY<BOOLEAN>, myOtherField TIMESTAMP(3)>

可在下文中找到所有预先定义好的数据类型。

Table API 中的数据类型

在定义 connector、catalog、用户自定义函数时，使用 JVM 相关 API 的用户可能会使用到 Table API 中基于 org.apache.flink.table.types.DataType 的一些实例。

数据类型 实例有两个职责：

• 作为逻辑类型的表现形式，定义 JVM 类语言或 Python 语言与 Table 生态系统的边界，而不是以具体的物理表现形式存在于数据的传输过程或存储中。
• 可选的: 在与其他 API 进行数据交换时，为 Planner 提供这些数据物理层面的相关提示。

对于基于 JVM 的语言，所有预定义的数据类型都可以在 org.apache.flink.table.api.DataTypes 下找到。

使用 Table API 编程时，建议使用星号引入所有相关依赖，以获得更流畅的 API 使用体验：

import static org.apache.flink.table.api.DataTypes.*;
DataType t = INTERVAL(DAY(), SECOND(3));

物理提示

在Table 生态系统中，当需要将 SQL 中的数据类型对应到实际编程语言中的数据类型时，就需要有物理提示。物理提示明确了对应过程中应该使用哪种数据格式。

比如，在 source 端产生数据时，可以规定：TIMESTAMP 的逻辑类型，在底层要使用 java.sql.Timestamp 这个类表示，而不是使用默认的 java.time.LocalDateTime 类。有了物理提示，可以帮助 Flink 运行时根据提供的类将数据转换为其内部数据格式。同样在 sink 端，定义好数据格式，以便能从 Flink 运行时获取、转换数据。

下面的例子展示了如何声明一个桥接转换类：

// 告诉 Flink 运行时使用 java.sql.Timestamp 处理数据，而不是 java.time.LocalDateTime
DataType t = DataTypes.TIMESTAMP(3).bridgedTo(java.sql.Timestamp.class);

// 告诉 Flink 运行时使用基本的 int 数组来处理数据，而不是用包装类 Integer 数组
DataType t = DataTypes.ARRAY(DataTypes.INT().notNull()).bridgedTo(int[].class);

注意 请记住，只有在扩展 API 时才需要使用到物理提示。使用预定义的 source、sink 以及 Flink 函数时，不需要用到物理提示。在使用 Table API 编写程序时，Flink 会忽略物理提示（例如 field.cast(TIMESTAMP(3).bridgedTo(Timestamp.class))）。

数据类型列表

下部分展示了所有预定义的数据类型

对于基于 JVM 的表 API，这些类型在 org.apache.flink.table.api.DataTypes 中也可用。

默认规划器支持以下一组 SQL 类型：

字符串

CHAR

定长字符串的数据类型。

声明

CHAR
CHAR(n)

可以使用 CHAR(n) 声明类型，其中 n 是代码点的数量。 n 的值必须介于 1 和 2,147,483,647（包括两者）之间。 如果未指定长度，则 n 等于 1。

VARCHAR/STRING

变长字符串的数据类型。

声明

VARCHAR
VARCHAR(n)

STRING

可以使用 VARCHAR(n) 声明类型，其中 n 是代码点的最大数量。 n 的值必须介于 1 和 2,147,483,647（包括两者）之间。 如果未指定长度，则 n 等于 1。

STRING 是 VARCHAR(2147483647) 的同义词。

二进制字符串

BINARY

固定长度二进制字符串（=字节序列）的数据类型。

声明

BINARY
BINARY(n)

可以使用 BINARY(n) 声明类型，其中 n 是字节数。 n 的值必须介于 1 和 2,147,483,647（包括两者）之间。 如果未指定长度，则 n 等于 1。

VARBINARY / BYTES

可变长度二进制字符串（=字节序列）的数据类型。

声明

VARBINARY
VARBINARY(n)

BYTES

可以使用 VARBINARY(n) 声明类型，其中 n 是最大字节数。 n 的值必须介于 1 和 2,147,483,647（包括两者）之间。 如果未指定长度，则 n 等于 1。

BYTES 是 VARBINARY(2147483647) 的同义词。

精确数字

DECIMAL

具有固定精度和小数位数的十进制数的数据类型。

声明

DECIMAL
DECIMAL(p)
DECIMAL(p, s)

DEC
DEC(p)
DEC(p, s)

NUMERIC
NUMERIC(p)
NUMERIC(p, s)

可以使用 DECIMAL(p, s) 声明类型，其中 p 是数字中的位数（精度），s 是数字中小数点右侧的位数（比例）。 p 的值必须介于 1 和 38（包括两者）之间。 s 的值必须介于 0 和 p（包括两者）之间。 p 的默认值为 10。s 的默认值为 0。

NUMERIC(p, s) 和 DEC(p, s) 是这种类型的同义词。

TYNYINT

1 字节有符号整数的数据类型，值从 -128 到 127。

声明

TINYINT

SMALLINT

2 字节有符号整数的数据类型，值从 -32,768 到 32,767。

声明

SMALLINT

INT

4 字节有符号整数的数据类型，值从 -2,147,483,648 到 2,147,483,647。

声明

INT

INTEGER

INTEGER 是这种类型的同义词。

BIGINT

8 字节有符号整数的数据类型，其值从 -9,223,372,036,854,775,808 到 9,223,372,036,854,775,807。

声明

BIGINT

近似数值

FLOAT

4 字节单精度浮点数的数据类型。

与 SQL 标准相比，类型不带参数。

声明

FLOAT

DOUBLE

8 字节双精度浮点数的数据类型。

声明

DOUBLE

DOUBLE PRECISION

DOUBLE PRECISION是这种类型的同义词。

Date和Time

DATE

日期数据类型，由年月日组成，取值范围为0000-01-01到9999-12-31。

与 SQL 标准相比，范围从 0000 年开始。

声明

DATE

TIME

不带时区的时间数据类型，由 hour:minute:second[.fractional] 组成，精度高达纳秒，值范围为 00:00:00.000000000 到 23:59:59.999999999。

与 SQL 标准相比，不支持闰秒（23:59:60 和 23:59:61），因为语义更接近 java.time.LocalTime。 未提供带时区的时间。

声明

TIME
TIME(p)

可以使用 TIME(p) 声明类型，其中 p 是小数秒的位数（精度）。 p 的值必须介于 0 和 9（包括两者）之间。 如果未指定精度，则 p 等于 0。

TIMESTAMP

不带时区的时间戳的数据类型，由年月日小时：分钟：秒[.fractional]组成，精度达到纳秒，值范围从 0000-01-01 00:00:00.000000000 到 9999-12-31 23 ：59：59.999999999。

声明

TIMESTAMP
TIMESTAMP(p)

TIMESTAMP WITHOUT TIME ZONE
TIMESTAMP(p) WITHOUT TIME ZONE

可以使用 TIMESTAMP(p) 声明类型，其中 p 是小数秒的位数（精度）。 p 的值必须介于 0 和 9（包括两者）之间。 如果未指定精度，则 p 等于 6。

TIMESTAMP(p) WITHOUT TIME ZONE 是这种类型的同义词。

TIMESTAMP WITH TIME ZONE

时间戳的数据类型，时区由年-月-日时:分:秒[.小数]区组成，精度高达纳秒，值范围从 0000-01-01 00:00:00.000000000 +14:59 到 9999 -12-31 23:59:59.999999999 -14:59。

与TIMESTAMP_LTZ相比，时区偏移信息物理存储在每个数据中。 它单独用于每次计算、可视化或与外部系统的通信。

声明

TIMESTAMP WITH TIME ZONE
TIMESTAMP(p) WITH TIME ZONE

TIMESTAMP_LTZ

具有本地时区的时间戳的数据类型，由年月日小时：分钟：秒 [.fractional] 区域组成，精度高达纳秒，值范围从 0000-01-01 00:00:00.000000000 +14:59 到 9999-12-31 23:59:59.999999999 -14:59。

不支持闰秒（23:59:60 和 23:59:61），因为语义更接近 java.time.OffsetDateTime。

与 TIMESTAMP WITH TIME ZONE 相比，时区偏移信息并没有物理存储在每个数据中。 相反，该类型采用表生态系统边缘的 UTC 时区中的 java.time.Instant 语义。 每个数据都在当前会话中配置的本地时区中进行解释，以进行计算和可视化。

此类型通过允许根据配置的会话时区解释 UTC 时间戳，填补了时区自由和时区强制时间戳类型之间的空白。

声明

TIMESTAMP_LTZ
TIMESTAMP_LTZ(p)

TIMESTAMP WITH LOCAL TIME ZONE
TIMESTAMP(p) WITH LOCAL TIME ZONE

可以使用 TIMESTAMP_LTZ(p) 声明该类型，其中 p 是小数秒的位数（精度）。 p 的值必须介于 0 和 9（包括两者）之间。 如果未指定精度，则 p 等于 6。

TIMESTAMP(p) WITH LOCAL TIME ZONE 是这种类型的同义词。

INTERVAL YEAR TO MOUTH

一组年月间隔类型的数据类型。

该类型必须参数化为以下分辨率之一：

• 年的间隔，
• 间隔数年至数月，
• 或几个月的间隔。

年-月的间隔由 +years-months 组成，其值范围从 -9999-11 到 +9999-11。

所有类型的分辨率的值表示都是相同的。 例如，50 个月的间隔始终以年到月的间隔格式（默认年份精度）表示：+04-02。

声明

INTERVAL YEAR
INTERVAL YEAR(p)
INTERVAL YEAR(p) TO MONTH
INTERVAL MONTH

可以使用上述组合声明类型，其中 p 是年的位数（年精度）。 p 的值必须介于 1 和 4 之间（包括两者）。 如果未指定年份精度，则 p 等于 2。

INTERVAL DAY TO SECOND

一组日间间隔类型的数据类型。

该类型必须参数化为以下分辨率之一，精度最高为纳秒：

• 天的间隔，

• 几天到几小时的间隔，

• 间隔几天到几分钟，

• 天到秒的间隔，

• 小时间隔，

• 间隔几小时到几分钟，

• 小时到秒的间隔，

• 分钟间隔，

• 分钟到秒的间隔，

• 或秒间隔。

白天时间间隔由 +days hours:months:seconds.fractional 组成，其值范围从 -999999 23:59:59.999999999 到 +999999 23:59:59.999999999。 所有类型的分辨率的值表示都是相同的。 例如，70 秒的间隔始终以天数到秒的间隔格式（默认精度）表示：+00 00:01:10.000000。

声明

INTERVAL DAY
INTERVAL DAY(p1)
INTERVAL DAY(p1) TO HOUR
INTERVAL DAY(p1) TO MINUTE
INTERVAL DAY(p1) TO SECOND(p2)
INTERVAL HOUR
INTERVAL HOUR TO MINUTE
INTERVAL HOUR TO SECOND(p2)
INTERVAL MINUTE
INTERVAL MINUTE TO SECOND(p2)
INTERVAL SECOND
INTERVAL SECOND(p2)

可以使用上述组合声明类型，其中 p1 是天的位数（天精度），p2 是小数秒的位数（小数精度）。 p1 的值必须介于 1 和 6 之间（包括两者）。 p2 的值必须介于 0 和 9（包括两者）之间。 如果未指定 p1，则默认等于 2。 如果未指定 p2，则默认等于 6。

结构化数据类型

ARRAY

具有相同子类型的元素数组的数据类型。

与 SQL 标准相比，数组的最大基数无法指定，而是固定为 2,147,483,647。 此外，支持任何有效类型作为子类型。

声明

ARRAY<t>
t ARRAY

可以使用 ARRAY 声明类型，其中 t 是包含元素的数据类型。

t ARRAY 是更接近 SQL 标准的同义词。 例如，INT ARRAY 等同于 ARRAY。

MAP

将键（包括 NULL）映射到值（包括 NULL）的关联数组的数据类型。 一个映射不能包含重复的键； 每个键最多可以映射到一个值。

没有元素类型的限制； 确保唯一性是用户的责任。

映射类型是 SQL 标准的扩展。

声明

MAP<kt, vt>

可以使用 MAP<kt, vt> 声明类型，其中 kt 是键元素的数据类型，vt 是值元素的数据类型。

MULTISET

多重集 (=bag) 的数据类型。 与集合不同，它允许每个具有共同子类型的元素有多个实例。 每个唯一值（包括 NULL）都映射到某个多重性。

没有元素类型的限制； 确保唯一性是用户的责任。

声明

MULTISET<t>
t MULTISET

可以使用 MULTISET 声明类型，其中 t 是包含元素的数据类型。

t MULTISET 是更接近 SQL 标准的同义词。 例如，INT MULTISET 等同于 MULTISET。

ROW

字段序列的数据类型。

字段由字段名称、字段类型和可选描述组成。 表的一行最具体的类型是行类型。 在这种情况下，行的每一列对应于与该列具有相同序号位置的行类型的字段。

与 SQL 标准相比，可选的字段描述简化了复杂结构的处理。

行类型类似于其他非标准兼容框架中已知的 STRUCT 类型。

声明

ROW<n0 t0, n1 t1, ...>
ROW<n0 t0 'd0', n1 t1 'd1', ...>

ROW(n0 t0, n1 t1, ...>
ROW(n0 t0 'd0', n1 t1 'd1', ...)

可以使用 ROW<n0 t0 ‘d0’, n1 t1 ‘d1’, …> 声明类型，其中 n 是字段的唯一名称，t 是字段的逻辑类型，d 是字段的描述 .

ROW(…) 是更接近 SQL 标准的同义词。 例如，ROW(myField INT, myOtherField BOOLEAN) 等同于 ROW<myField INT, myOtherField BOOLEAN>。

用户定义数据类型

尚未完全支持用户定义的数据类型。 它们目前（从 Flink 1.11 开始）仅在函数的参数和返回类型中作为未注册的结构化类型公开。

结构化类型类似于面向对象编程语言中的对象。 它包含零个、一个或多个属性。 每个属性都由名称和类型组成。

有两种结构化类型：

• 存储在目录中并由目录标识符标识的类型（如 cat.db.MyType）。 这些等同于结构化类型的 SQL 标准定义。
• 由实现类（如 com.myorg.model.MyType）标识的匿名定义的未注册类型（通常是反射提取的）。 这些在以编程方式定义表程序时很有用。 它们允许重用现有的 JVM 类，而无需再次手动定义数据类型的模式。

注册结构类型

当前，不支持已注册的结构化类型。 因此，它们不能存储在目录中或在 CREATE TABLE DDL 中引用。

未注册的结构类型

可以使用自动反射提取从常规 POJO（普通旧 Java 对象）创建未注册的结构化类型。

结构化类型的实现类必须满足以下要求：

• 该类必须是全局可访问的，这意味着它必须声明为公共的、静态的，而不是抽象的。

• 该类必须提供带零参数的默认构造函数或分配所有字段的完整构造函数。

• 类的所有字段必须可通过公共声明或遵循通用编码风格（如 getField()、isField()、field()）的 getter 读取。

• 类的所有字段必须可以通过公开声明、完全分配的构造函数或遵循常见编码风格（例如 setField(…)、field(…)）的 setter 来写入。

• 所有字段都必须通过反射提取隐式或显式使用 @DataTypeHint 注释映射到数据类型。

• 声明为静态或瞬态的字段将被忽略。

只要字段类型不（传递地）引用自身，反射提取就支持字段的任意嵌套。

声明的字段类（例如 public int age;）必须包含在为本文档中的每种数据类型定义的支持的 JVM 桥接类列表中（例如 java.lang.Integer 或 int for INT）。

对于某些类，需要注释才能将类映射到数据类型（例如 @DataTypeHint("DECIMAL(10, 2)") 为 java.math.BigDecimal 分配固定的精度和比例）。

声明

class User {

    // extract fields automatically
    public int age;
    public String name;

    // enrich the extraction with precision information
    public @DataTypeHint("DECIMAL(10, 2)") BigDecimal totalBalance;

    // enrich the extraction with forcing using RAW types
    public @DataTypeHint("RAW") Class<?> modelClass;
}

DataTypes.of(User.class);

桥接到 JVM 类型

其他数据类型

BOOLEAN

具有（可能）三值逻辑 TRUE、FALSE 和 UNKNOWN 的布尔数据类型。

BOOLEAN

RAW

任意序列化类型的数据类型。 这种类型是表生态系统中的黑盒，仅在边缘反序列化。

原始类型是 SQL 标准的扩展。

声明

RAW('class', 'snapshot')

可以使用 RAW(‘class’, ‘snapshot’) 声明类型，其中 class 是原始类，snapshot 是 Base64 编码的序列化 TypeSerializerSnapshot。 通常，类型字符串不是直接声明的，而是在持久化类型时生成的。

在 API 中，可以通过直接提供 Class + TypeSerializer 或通过传递 Class 并让框架从那里提取 Class + TypeSerializer 来声明 RAW 类型。

NULL

表示无类型 NULL 值的数据类型。

null 类型是对 SQL 标准的扩展。 null 类型除了 NULL 没有其他值，因此，它可以转换为类似于 JVM 语义的任何可空类型。

此类型有助于在使用 NULL 文字的 API 调用中表示未知类型，以及桥接到定义此类类型的格式，例如 JSON 或 Avro。

这种类型在实践中不是很有用，这里只是为了完整性而提到。

声明

NULL

转换

Flink Table API 和 SQL 可以在定义的输入类型和目标类型之间进行转换。 尽管一些强制转换操作总是可以成功，而不管输入值如何，但其他操作可能会在运行时失败（即无法为目标类型创建值）。 例如，总是可以将 INT 转换为 STRING，但不能总是将 STRING 转换为 INT。

在规划阶段，查询验证器会拒绝带有 ValidationException 的无效类型对的查询，例如 尝试将 TIMESTAMP 转换为 INTERVAL 时。 可能在运行时失败的有效类型对将被查询验证器接受，但需要用户正确处理失败。

在 Flink Table API 和 SQL 中，可以使用以下两个内置函数之一来执行转换：

• CAST：SQL 标准定义的常规转换函数。 如果转换操作容易出错并且提供的输入无效，则作业可能会失败。 类型推断将保留输入类型的可空性。
• TRY_CAST：常规转换函数的扩展，在转换操作失败时返回 NULL。 它的返回类型总是可以为空的。

例如：

CAST('42' AS INT) --- returns 42 of type INT NOT NULL
CAST(NULL AS VARCHAR) --- returns NULL of type VARCHAR
CAST('non-number' AS INT) --- throws an exception and fails the job

TRY_CAST('42' AS INT) --- returns 42 of type INT
TRY_CAST(NULL AS VARCHAR) --- returns NULL of type VARCHAR
TRY_CAST('non-number' AS INT) --- returns NULL of type INT
COALESCE(TRY_CAST('non-number' AS INT), 0) --- returns 0 of type INT NOT NULL

下面的矩阵描述了支持的转换对，其中“Y”表示支持，“！” 表示容易出错，“N”表示不受支持：

注意：

1. 所有转换为恒定长度或可变长度的类型也将根据类型定义进行修剪和填充。
2. 必须使用 TO_TIMESTAMP 和 TO_TIMESTAMP_LTZ 而不是 CAST/TRY_CAST。
3. 如果支持子类型对，则支持。 当且仅当子类型对是易错的。
4. 如果 RAW 类和序列化程序相等，则支持。
5. 支持的 iff INTERVAL 是 MONTH TO YEAR 范围。
6. 支持的 iff INTERVAL 是 DAY TO TIME 范围。

另请注意，无论使用的函数是 CAST 还是 TRY_CAST，NULL 值的转换总是返回 NULL。

数据类型提取

在 API 的很多地方，Flink 都尝试使用反射自动从类信息中提取数据类型，以避免重复的手动 schema 工作。 但是，反射式提取数据类型并不总是成功，因为可能会丢失逻辑信息。 因此，可能需要在类或字段声明附近添加附加信息以支持提取逻辑。

下表列出了无需进一步信息即可隐式映射到数据类型的类。

如果你打算在 Scala 中实现类，建议使用装箱类型（例如 java.lang.Integer）而不是 Scala 的原语。 Scala 的原语（例如 Int 或 Double）被编译为 JVM 原语（例如 int/double）并产生 NOT NULL 语义，如下表所示。 此外，在泛型中使用的 Scala 基元（例如 java.util.Map[Int, Double]）在编译期间被擦除并导致类信息类似于 java.util.Map[java.lang.Object, java.lang.Object ].

本文档中提到的其他 JVM 桥接类需要 @DataTypeHint 注释。

数据类型提示可以参数化或替换单个函数参数和返回类型、结构化类或结构化类的字段的默认提取逻辑。 实现者可以通过声明 @DataTypeHint 注释来选择默认提取逻辑应该修改到什么程度。

@DataTypeHint 注释提供了一组可选的提示参数。 以下示例显示了其中一些参数。 更多信息可以在注释类的文档中找到。

import org.apache.flink.table.annotation.DataTypeHint;

class User {

    // defines an INT data type with a default conversion class `java.lang.Integer`
    public @DataTypeHint("INT") Object o;

    // defines a TIMESTAMP data type of millisecond precision with an explicit conversion class
    public @DataTypeHint(value = "TIMESTAMP(3)", bridgedTo = java.sql.Timestamp.class) Object o;

    // enrich the extraction with forcing using a RAW type
    public @DataTypeHint("RAW") Class<?> modelClass;

    // defines that all occurrences of java.math.BigDecimal (also in nested fields) will be
    // extracted as DECIMAL(12, 2)
    public @DataTypeHint(defaultDecimalPrecision = 12, defaultDecimalScale = 2) AccountStatement stmt;

    // defines that whenever a type cannot be mapped to a data type, instead of throwing
    // an exception, always treat it as a RAW type
    public @DataTypeHint(allowRawGlobally = HintFlag.TRUE) ComplexModel model;
}