Flink流批一体计算(21):Flink SQL之Flink DDL

2023-11-29 10:30

本文主要是介绍Flink流批一体计算(21):Flink SQL之Flink DDL,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

目录

执行 CREATE 语句

Python脚本

Java代码

SQL语句

列定义

物理/常规列

元数据列

计算列

WATERMARK

PRIMARY KEY

PARTITIONED BY

AS select_statement


Flink SQL是为了简化计算模型、降低您使用Flink门槛而设计的一套符合标准SQL语义的开发语言。

执行 CREATE 语句

Python脚本
table_env = TableEnvironment.create(...)# 对已经注册的表进行 SQL 查询# 注册名为 “Orders” 的表table_env.execute_sql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");# 在表上执行 SQL 查询,并把得到的结果作为一个新的表result = table_env.sql_query("SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");# 对已注册的表进行 INSERT 操作# 注册 TableSinktable_env.execute_sql("CREATE TABLE RubberOrders(product STRING, amount INT) WITH (...)")# 在表上执行 INSERT 语句并向 TableSink 发出结果table_env \.execute_sql("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")

Java代码
Environmentsettings settings = Environmentsettings.newInstance()...TableEnvironment tableEnv = TableEnvironment.create(settings);// 对已注册的表进行 SQL 查询// 注册名为 “Orders” 的表tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");// 在表上执行 SQL 查询,并把得到的结果作为一个新的表Table result = tableEnv.sqlQuery("SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");// 对已注册的表进行 INSERT 操作// 注册 TableSinktableEnv.executeSql("CREATE TABLE RubberOrders(product STRING, amount INT) WITH (...)");// 在表上执行 INSERT 语句并向 TableSink 发出结果tableEnv.executeSql("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");

SQL语句
Flink SQL> CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...);[INFO] Table has been created.Flink SQL> CREATE TABLE RubberOrders (product STRING, amount INT) WITH (...);[INFO] Table has been created.Flink SQL> INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%';[INFO] Submitting SQL update statement to the cluster...

列定义

物理/常规列

物理列是数据库中已知的常规列。它们定义物理数据中字段的名称、类型和顺序。因此,物理列表示从外部系统读取和写入的有效数据。其他类型的列可以在物理列之间声明,但不会影响最终的物理模式。

以下语句创建一个仅包含常规列的表:

CREATE TABLE MyTable (

  `user_id` BIGINT,

  `name` STRING

) WITH (

  ...

);

元数据列

元数据列是SQL标准的扩展,元数据列由metadata关键字指示。例如,可以使用元数据列从Kafka记录读取时间戳,并将时间戳写入Kafka,以进行基于时间的操作。连接器和格式文档列出了每个组件的可用元数据字段。但是,在表的模式中声明元数据列是可选的。

以下语句创建一个表,其中包含引用元数据字段时间戳的附加元数据列:

CREATE TABLE MyTable (

  `user_id` BIGINT,

  `name` STRING,

  `record_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'    -- reads and writes a Kafka record's timestamp

) WITH (

  'connector' = 'kafka'

  ...

);

每个元数据字段都由基于字符串的键标识,并具有文档化的数据类型。例如,Kafka连接器公开了一个元数据字段,该字段具有键时间戳和数据类型timestamp_LTZ3),可用于读取和写入记录。

在上面的示例中,元数据列record_time成为表模式的一部分,可以像常规列一样进行转换和存储,为方便起见,如果列名应用作标识元数据键,则可以省略FROM子句。

CREATE TABLE MyTable (

  `user_id` BIGINT,

  `name` STRING,

  `timestamp` TIMESTAMP_LTZ(3) METADATA    -- use column name as metadata key

) WITH (

  'connector' = 'kafka'

  ...

);

计算列

计算列是使用语法column_name AS Computed_column_expression生成的虚拟列。

计算列计算可以引用同一表中声明的其他列的表达式。可以访问物理列和元数据列。列本身不物理存储在表中。列的数据类型是从给定表达式自动派生的,不必手动声明。

例如,计算列可以定义为

CREATE TABLE MyTable (

  `user_id` BIGINT,

  `price` DOUBLE,

  `quantity` DOUBLE,

  `cost` AS price * quanitity,  -- evaluate expression and supply the result to queries

) WITH (

  'connector' = 'kafka'

  ...

);

Flink中通常使用计算列来定义CREATETABLE语句中的时间属性。

使用系统的PROCTIME()函数,可以通过proc AS PROCIME()轻松定义processing time属性。

Event time属性时间戳可以在WATERMARK声明之前预处理。例如,如果原始字段不是TIMESTAMP3)类型或嵌套在JSON字符串中,则可以使用计算列。

WATERMARK

WATERMARK 定义了表的Event time属性,其形式为 WATERMARK FOR rowtime_column_name AS watermark_strategy_expression 

rowtime_column_name 把一个现有的列定义为一个为表标记event time的属性。该列的类型必须为 TIMESTAMP(3),且是 schema 中的顶层列,它也可以是一个计算列。

watermark_strategy_expression 定义了 watermark 的生成策略。它允许使用包括计算列在内的任意非查询表达式来计算 watermark ;表达式的返回类型必须是 TIMESTAMP(3),表示了从 Epoch 以来的经过的时间。 返回的 watermark 只有当其不为空且其值大于之前发出的本地 watermark 时才会被发出(以保证 watermark 递增)。每条记录的 watermark 生成表达式计算都会由框架完成。 框架会定期发出所生成的最大的 watermark ,如果当前 watermark 仍然与前一个 watermark 相同、为空、或返回的 watermark 的值小于最后一个发出的 watermark ,则新的 watermark 不会被发出。 Watermark 根据 pipeline.auto-watermark-interval 中所配置的间隔发出。 watermark 的间隔是 0ms ,那么每条记录都会产生一个 watermark,且 watermark 会在不为空并大于上一个发出的 watermark 时发出。

使用事件时间语义时,表必须包含事件时间属性和 watermark 策略。

Flink 提供了几种常用的 watermark 策略。

  • 严格递增时间戳: WATERMARK FOR rowtime_column AS rowtime_column

发出到目前为止已观察到的最大时间戳的 watermark ,时间戳大于最大时间戳的行被认为没有迟到。

  • 递增时间戳: WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '0.001' SECOND

发出到目前为止已观察到的最大时间戳减 1 watermark ,时间戳大于或等于最大时间戳的行被认为没有迟到。

  • 有界乱序时间戳: WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL 'string' timeUnit

发出到目前为止已观察到的最大时间戳减去指定延迟的 watermark ,例如, WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '5' SECOND 是一个 5 秒延迟的 watermark 策略。

CREATE TABLE Orders (

    `user` BIGINT,

    product STRING,

    order_time TIMESTAMP(3),

    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND

) WITH (......);

PRIMARY KEY

主键用作 Flink 优化的一种提示信息。主键限制表明一张表或视图的某个(些)列是唯一的并且不包含 Null 值。 主键声明的列都是非 nullable 的。因此主键可以被用作表行级别的唯一标识。

主键可以和列的定义一起声明,也可以独立声明为表的限制属性,不管是哪种方式,主键都不可以重复定义,否则 Flink 会报错。

有效性检查

SQL 标准主键限制可以有两种模式:ENFORCED 或者 NOT ENFORCED 它申明了是否输入/出数据会做合法性检查(是否唯一)。Flink 不存储数据因此只支持 NOT ENFORCED 模式,即不做检查,用户需要自己保证唯一性。

Flink 假设声明了主键的列都是不包含 Null 值的,Connector 在处理数据时需要自己保证语义正确。

Notes:  CREATE TABLE 语句中,创建主键会修改列的 nullable 属性,主键声明的列默认都是非 Nullable 的。

PARTITIONED BY

根据指定的列对已经创建的表进行分区。若表使用 filesystem sink ,则将会为每个分区创建一个目录。

AS select_statement

表也可以通过一个 CTAS 语句中的查询结果来创建并填充数据,CTAS 是一种简单、快捷的创建表并插入数据的方法。

CTAS 有两个部分,SELECT 部分可以是 Flink SQL 支持的任何 SELECT 查询 CREATE 部分从 SELECT 查询中获取列信息,并创建目标表。  CREATE TABLE 类似,CTAS 要求必须在目标表的 WITH 子句中指定必填的表属性。

CTAS 的建表操作需要依赖目标 Catalog。比如,Hive Catalog 会自动在 Hive 中创建物理表。但是基于内存的 Catalog 只会将表的元信息注册在执行 SQL Client 的内存中。

CREATE TABLE my_ctas_table

WITH (

    'connector' = 'kafka',

    ...

)

AS SELECT id, name, age FROM source_table WHERE mod(id, 10) = 0;

注意 CTAS 有如下约束:

  • 暂不支持创建临时表。
  • 暂不支持指定列信息。
  • 暂不支持指定 Watermark。
  • 暂不支持创建分区表。
  • 暂不支持主键约束。

注意 目前,CTAS 创建的目标表是非原子性的,如果在向表中插入数据时发生错误,该表不会被自动删除。

这篇关于Flink流批一体计算(21):Flink SQL之Flink DDL的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/432479

相关文章

SQL中的外键约束

外键约束用于表示两张表中的指标连接关系。外键约束的作用主要有以下三点: 1.确保子表中的某个字段(外键)只能引用父表中的有效记录2.主表中的列被删除时,子表中的关联列也会被删除3.主表中的列更新时,子表中的关联元素也会被更新 子表中的元素指向主表 以下是一个外键约束的实例展示

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后,随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步,我们目前的解决方案是建立数据中间表。把需要检索的业务数据,统一放到一张M

如何去写一手好SQL

MySQL性能 最大数据量 抛开数据量和并发数,谈性能都是耍流氓。MySQL没有限制单表最大记录数,它取决于操作系统对文件大小的限制。 《阿里巴巴Java开发手册》提出单表行数超过500万行或者单表容量超过2GB,才推荐分库分表。性能由综合因素决定,抛开业务复杂度,影响程度依次是硬件配置、MySQL配置、数据表设计、索引优化。500万这个值仅供参考,并非铁律。 博主曾经操作过超过4亿行数据

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题,本文将一起从代码层去分析为什么没有建立索引? 开源ERP项目地址:https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置

MySQL数据库宕机,启动不起来,教你一招搞定!

作者介绍:老苏,10余年DBA工作运维经验,擅长Oracle、MySQL、PG、Mongodb数据库运维(如安装迁移,性能优化、故障应急处理等)公众号:老苏畅谈运维欢迎关注本人公众号,更多精彩与您分享。 MySQL数据库宕机,数据页损坏问题,启动不起来,该如何排查和解决,本文将为你说明具体的排查过程。 查看MySQL error日志 查看 MySQL error日志,排查哪个表(表空间

MySQL高性能优化规范

前言:      笔者最近上班途中突然想丰富下自己的数据库优化技能。于是在查阅了多篇文章后,总结出了这篇! 数据库命令规范 所有数据库对象名称必须使用小写字母并用下划线分割 所有数据库对象名称禁止使用mysql保留关键字(如果表名中包含关键字查询时,需要将其用单引号括起来) 数据库对象的命名要能做到见名识意,并且最后不要超过32个字符 临时库表必须以tmp_为前缀并以日期为后缀,备份

poj 1113 凸包+简单几何计算

题意: 给N个平面上的点,现在要在离点外L米处建城墙,使得城墙把所有点都包含进去且城墙的长度最短。 解析: 韬哥出的某次训练赛上A出的第一道计算几何,算是大水题吧。 用convexhull算法把凸包求出来,然后加加减减就A了。 计算见下图: 好久没玩画图了啊好开心。 代码: #include <iostream>#include <cstdio>#inclu

uva 1342 欧拉定理(计算几何模板)

题意: 给几个点,把这几个点用直线连起来,求这些直线把平面分成了几个。 解析: 欧拉定理: 顶点数 + 面数 - 边数= 2。 代码: #include <iostream>#include <cstdio>#include <cstdlib>#include <algorithm>#include <cstring>#include <cmath>#inc

uva 11178 计算集合模板题

题意: 求三角形行三个角三等分点射线交出的内三角形坐标。 代码: #include <iostream>#include <cstdio>#include <cstdlib>#include <algorithm>#include <cstring>#include <cmath>#include <stack>#include <vector>#include <

[MySQL表的增删改查-进阶]

🌈个人主页:努力学编程’ ⛅个人推荐: c语言从初阶到进阶 JavaEE详解 数据结构 ⚡学好数据结构,刷题刻不容缓:点击一起刷题 🌙心灵鸡汤:总有人要赢,为什么不能是我呢 💻💻💻数据库约束 🔭🔭🔭约束类型 not null: 指示某列不能存储 NULL 值unique: 保证某列的每行必须有唯一的值default: 规定没有给列赋值时的默认值.primary key: