flink-38 Flink SQL DDL

可以修改数据库属性

ALTER DATABASE [catalog_name.]db_name SET (key1=value1, ...)

删除数据库

DROP DATABASE [IF EXISTS] [catalog_name.]db_name [(RESTRICT | CASCADE)]

RESTRICT 删除非空数据库触发异常默认启用
CASCADE 删除非空数据库也会删除所有相关的表和函数

physical_column_definition
- 物理列时数据库中所说的常规列。
metadata_column_definition
- 元数据列是SQL标准的扩展，允许访问数据源本身具有的元数据
- 使用METADATA关键字标识

可以使用元数据列从kafka记录中读取和写入时间戳

CREATE TABLE MyTable (
    user_id STRING,
    name STRING,
    // 获取kafka的元数据
    record_time TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
    'connector' = 'kafka'
)

也可以强制转换类型

timestamp BIGINT METADATA

多数情况下外部系统提供的只读元数据字段比可写字段多，因此可以使用VIRTUAL关键字排除元数据列的持久化

timestamp BIGINT METADATA
offset BIGINT METADATA VIRTUAL

computed_column_definition

计算列，使用语法column_name AS 计算表达式

cost AS price * quanitity

定义watermark
- 严格升序： WATERMARK FOR rowtime_column AS rowtime_column
  - flink任务认为时间戳只会越来越大，也不存在相等的情况，只要相等或者小于之前的，就是迟到的数据
- 递增：WATERMARK FOR rowtime_column AS rowtime_column – INTERVAL ‘0.001’ SECOND
  - 一般基本不用这种方式如果设置此类，则允许有相同的时间戳出现
- 有界无序：WATERMARK FOR rowtime_column AS rowtime_column – INTERVAL ‘string’ timeUint
  - 用于设置最大乱序时间
PRIMARY KEY
- 主键约束表明表中的一列或一组列是唯一的
- 并且他们不包含NULL值
- 支持持not enforce

PRIMARY KEY(user_id) not enforce

PARTITIONED BY
WITH
- 用于创建表的表属性
- 用于指定外部存储系统的元数据信息
- 格式：key=val 字符串类型
LIKE
- 基于现有的表定义创建表
- 用户可以扩展表和排除表的某些部分
AS
- CREATE TABLE … AS SELECT 语句
- 暂不支持创建临时表
- 不支持显示列
- 不支持指定显示水印
- 不支持创建分区表
- 不支持指定主键约束

创建简单表

CREATE TABLE test (
    id INT,
    ts BIGINT,
    vc INT
) WITH (
    'connector' = 'print'
);

扩展表

CREATE TABLE test1 (`value` STRING) LIKE test;

查看表

SHOW TABLES [(FROM|IN) [catalog_name.]database_name] [NOT][LIKE] <sql匹配>

Blog

flink-38 Flink SQL DDL