分享好友 站长动态首页 网站导航

Flink SQL 知其所以然:SQL 数据类型大全!

2022-05-19 10:00 · 头闻号数据库

SQL 数据类型

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

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

总共包含 3 部分:

一、原子数据类型

1、字符串类型:

2、二进制字符串类型:

3、 精确数值类型:

4、有损精度数值类型:

5、布尔类型:BOOLEAN。

6、NULL 类型:NULL。

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

8、日期、时间类型:

CREATE TABLE sink_table (
result_interval_year TIMESTAMP(3),
result_interval_year_p TIMESTAMP(3),
result_interval_year_p_to_month TIMESTAMP(3),
result_interval_month TIMESTAMP(3),
result_interval_day TIMESTAMP(3),
result_interval_day_p1 TIMESTAMP(3),
result_interval_day_p1_to_hour TIMESTAMP(3),
result_interval_day_p1_to_minute TIMESTAMP(3),
result_interval_day_p1_to_second_p2 TIMESTAMP(3),
result_interval_hour TIMESTAMP(3),
result_interval_hour_to_minute TIMESTAMP(3),
result_interval_hour_to_second TIMESTAMP(3),
result_interval_minute TIMESTAMP(3),
result_interval_minute_to_second_p2 TIMESTAMP(3),
result_interval_second TIMESTAMP(3),
result_interval_second_p2 TIMESTAMP(3)
) WITH (
'connector' = 'print'
);
INSERT INTO sink_table
SELECT
-- Flink SQL 支持的所有 INTERVAL 子句如下,总体可以分为 `年-月`、`日-小时-秒` 两种

-- 1. 年-月。取值范围为 [-9999-11, +9999-11],其中 p 是指有效位数,取值范围 [1, 4],默认值为 2。比如如果值为 1000,但是 p = 2,则会直接报错。
-- INTERVAL YEAR
f1 + INTERVAL '10' YEAR as result_interval_year
-- INTERVAL YEAR(p)
, f1 + INTERVAL '100' YEAR(3) as result_interval_year_p
-- INTERVAL YEAR(p) TO MONTH
, f1 + INTERVAL '10-03' YEAR(3) TO MonTH as result_interval_year_p_to_month
-- INTERVAL MONTH
, f1 + INTERVAL '13' MonTH as result_interval_month

-- 2. 日-小时-秒。取值范围为 [-999999 23:59:59.999999999, +999999 23:59:59.999999999],其中 p1p2 都是有效位数,p1 取值范围 [1, 6],默认值为 2;p2 取值范围 [0, 9],默认值为 6
-- INTERVAL DAY
, f1 + INTERVAL '10' DAY as result_interval_day
-- INTERVAL DAY(p1)
, f1 + INTERVAL '100' DAY(3) as result_interval_day_p1
-- INTERVAL DAY(p1) TO HOUR
, f1 + INTERVAL '10 03' DAY(3) TO HOUR as result_interval_day_p1_to_hour
-- INTERVAL DAY(p1) TO MINUTE
, f1 + INTERVAL '10 03:12' DAY(3) TO MINUTE as result_interval_day_p1_to_minute
-- INTERVAL DAY(p1) TO SECOND(p2)
, f1 + INTERVAL '10 00:00:00.004' DAY TO SECOND(3) as result_interval_day_p1_to_second_p2
-- INTERVAL HOUR
, f1 + INTERVAL '10' HOUR as result_interval_hour
-- INTERVAL HOUR TO MINUTE
, f1 + INTERVAL '10:03' HOUR TO MINUTE as result_interval_hour_to_minute
-- INTERVAL HOUR TO SECOND(p2)
, f1 + INTERVAL '00:00:00.004' HOUR TO SECOND(3) as result_interval_hour_to_second
-- INTERVAL MINUTE
, f1 + INTERVAL '10' MINUTE as result_interval_minute
-- INTERVAL MINUTE TO SECOND(p2)
, f1 + INTERVAL '05:05.006' MINUTE TO SECOND(3) as result_interval_minute_to_second_p2
-- INTERVAL SECOND
, f1 + INTERVAL '3' SECOND as result_interval_second
-- INTERVAL SECOND(p2)
, f1 + INTERVAL '300' SECOND(3) as result_interval_second_p2
FROM (SELECT TO_TIMESTAMP_LTZ(1640966476500, 3) as f1)

二、复合数据类型

  1. 数组类型:ARRAY、t ARRAY。数组最大长度为 2,147,483,647。t 代表数组内的数据类型。举例 ARRAY、ARRAY,其等同于 INT ARRAY、STRING ARRAY。
  2. Map 类型:MAP。Map 类型就和 Java 中的 Map 类型一样,key 是没有重复的。举例 Map、Map。
  3. 集合类型:MULTISET、t MULTISET。就和 Java 中的 List 类型,一样,运行重复的数据。举例 MULTISET,其等同于 INT MULTISET。
  4. 对象类型: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 进行定义,只支持作为函数的输入输出参数。如下案例:

第一步,自定义数据类型

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

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

public class UserScalarFunction extends ScalarFunction {
// 1. 自定义数据类型作为输出参数
public User eval(long i) {
if (i > 0 && i <= 5) {
User u = new User();
u.age = (int) i;
u.name = "name1";
u.totalBalance = new BigDecimal(1.1d);
return u;
} else {
User u = new User();
u.age = (int) i;
u.name = "name2";
u.totalBalance = new BigDecimal(2.2d);
return u;
}
}
// 2. 自定义数据类型作为输入参数
public String eval(User i) {
if (i.age > 0 && i.age <= 5) {
User u = new User();
u.age = 1;
u.name = "name1";
u.totalBalance = new BigDecimal(1.1d);
return u.name;
} else {
User u = new User();
u.age = 2;
u.name = "name2";
u.totalBalance = new BigDecimal(2.2d);
return u.name;
}
}
}

第三步,在 Flink SQL 中使用

-- 1. 创建 UDF
CREATE FUNCTION user_scalar_func AS 'flink.examples.sql._12_data_type._02_user_defined.UserScalarFunction';
-- 2. 创建数据源表
CREATE TABLE source_table (
user_id BIGINT NOT NULL COMMENT '用户 id'
) WITH (
'connector' = 'datagen',
'rows-per-second' = '1',
'fields.user_id.min' = '1',
'fields.user_id.max' = '10'
);

-- 3. 创建数据汇表
CREATE TABLE sink_table (
result_row_1 ROW<age INT, name STRING, totalBalance DECIMAL(10, 2)>,
result_row_2 STRING
) WITH (
'connector' = 'print'
);
-- 4. SQL 查询语句
INSERT INTO sink_table
select
-- 4.a. 用户自定义类型作为输出
user_scalar_func(user_id) as result_row_1,
-- 4.b. 用户自定义类型作为输出及输入
user_scalar_func(user_scalar_func(user_id)) as result_row_2
from source_table;
-- 5. 查询结果
+I[+I[9, name2, 2.20], name2]
+I[+I[1, name1, 1.10], name1]
+I[+I[5, name1, 1.10], name1]

免责声明:本平台仅供信息发布交流之途,请谨慎判断信息真伪。如遇虚假诈骗信息,请立即举报

举报
反对 0
打赏 0
更多相关文章

评论

0

收藏

点赞