2022-06-21-Flink-49(一. SQL手册)

作者: 冰菓_ | 来源:发表于2022-06-21 22:38 被阅读0次

1. 动态表

SQL	流计算
表是有界数据集,计算过程中,一旦确认了就不会改变	流是无界数据集,数据持续产生,计算过程中,数据集持续在变化
对批处理数据执行的查询结果可以访问完整的输入数据	流式查询在启动时无法访问所有数据,必须等待数据流入
对有界数据集的计算最终会结束,得到计算结果	流式会持续收到数据,不断更新其计算结果,不会结束

传统SQL比较成熟,且应用广泛,如果能够在流上使用SQL进行开发,那么会带来极大的便利,将SQL应用于流计算,需要将流和表两个差异比较大的概念进行融合.如果把有界数据集当成表,那么无界数据集就是一个随着时间变化持续写入数据的表.

2. ⼀个 SQL\Table API 任务的代码结构

public class tableDemo1 {

    public static void main(String[] args) {
        EnvironmentSettings setting  = EnvironmentSettings.newInstance()
                .inStreamingMode() //声明成流任务
                // .inBatchMode() // 声明成批任务
                .build();

        TableEnvironment tableEnv = TableEnvironment.create(setting);
        //创建一个输入表
        tableEnv.executeSql("CREATE TABLE input_table(user_name STRING ,url STRING, ts BIGINT) WITH ( 'connector' = 'filesystem', 'path' = 'src/main/resources/c.txt', 'format' = 'csv' )");
        ////创建一个输出表
        tableEnv.executeSql("CREATE TABLE output_table(user_name STRING ,url STRING) WITH ( 'connector' = 'filesystem', 'path' = 'output/2022-06-21', 'format' = 'csv' )");

        //使用TABLE-API
        Table table_API = tableEnv.from("input_table").where($("user_name").isEqual("小米")).select($("user_name"), $("url"));
        //使用SQL
        Table table_SQL = tableEnv.sqlQuery("SELECT user_name, url FROM input_table WHERE user_name = '小米' ");
        //将数据写入输出表中
        table_API.executeInsert("output_table");
        table_SQL.executeInsert("output_table");

    }
}

SQL上下文:TableEnvironment

在内部catalog中注册表,catalog可以理解为flink的metastore,类似Hive中的metastore对Hive的地位
注册外部catalog
执行SQL查询
注册用户定义（标量，表或聚合）函数
将DataStream或DataSet(废弃)转换为表
持有对ExecutionEnvironment或StreamExecutionEnvironment的引用

SQL中的表概念

一个表的全名(标识)会由三个部分组成:catalog 名称.数据库名称.表名称,如果catalog名称或者数据库名称没有指名,就会使用当前默认值default,下面这个SQL创建的table的全名就是default.nono.output_table

tableEnv.executeSql("CREATE TABLE nono.output_table(user_name STRING ,url STRING) WITH ( 'connector' = 'filesystem', 'path' = 'output/2022-06-21', 'format' = 'csv' )");

表可以是常规的外部表,也可以是虚拟的视图

外部表:描述的是外部数据,例如是文件,消息队列等
视图view:从已经存在的表中创建,视图一般是一个SQL逻辑的查询结果,对比到离线的hive SQL 中,在离线的场景中的view也都是从已经有的表中去创建

一个SQL查询案例

聊聊flink 1.11 中的随机数据生成器-DataGen connector

public class tableDemo2 {

    public static void main(String[] args) {

        EnvironmentSettings setting  = EnvironmentSettings.newInstance()
                .inStreamingMode() //声明成流任务
                // .inBatchMode() // 声明成批任务
                .build();

        TableEnvironment tableEnv = TableEnvironment.create(setting);
        //创建随机mock数据源
        tableEnv.executeSql("CREATE TABLE sku_price_table( tag STRING ,sku_id INT ,price INT,ts AS localtimestamp, WATERMARK FOR ts AS ts ) " +
                                     " WITH( 'connector' = 'datagen',  " +
                                             " 'rows-per-second'='1',  " +
                                             " 'fields.sku_id.kind'='sequence', " +
                                             " 'fields.sku_id.start'='1'," +
                                             " 'fields.sku_id.end'='1000'," +
                                             " 'fields.tag.length'='1'," +
                                             " 'fields.price.min'='1'," +
                                             " 'fields.price.max'='1000')");

        //创建到kafka的输入表
        tableEnv.executeSql("CREATE TABLE sink_tag_agg_Table (" +
                " tag STRING, " +
                " count_tag BIGINT, " +
                " sum_tag BIGINT , " +
                " max_tag BIGINT, " +
                " PRIMARY KEY (`tag`) NOT ENFORCED" +
                ") WITH (" +
                " 'connector' = 'upsert-kafka', " +
                " 'topic' = 'topic05', " +
                " 'properties.bootstrap.servers' = '43.142.80.86:9093', " +
                " 'key.format' = 'json' ," +
                " 'value.format' = 'json'" +
                ")");

    tableEnv.executeSql("INSERT INTO  sink_tag_agg_Table SELECT tag, count(1) as count_tag, sum(price) as sum_tag,max(price) as max_tag FROM sku_price_table GROUP BY tag");
    }
}

bug:Could not find any factory for identifier 'json' that implements
缺少依赖,加入json依赖

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
        </dependency>

SQL 与 DataStream API 的转换

SQL实现不了的功能DataStream API来实现喽,如下实现一个log告警功能

public class tableDemo3 {

    public static void main(String[] args) throws Exception {
        LocalStreamEnvironment env= StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //创建随机mock数据源
        tableEnv.executeSql("CREATE TABLE tag_money_table( tag STRING ,id INT ,money INT,ts AS CAST(CURRENT_TIMESTAMP AS timestamp_LTZ(3)), WATERMARK FOR ts AS ts - INTERVAL '1' SECOND) " +
                " WITH( 'connector' = 'datagen',  " +
                " 'rows-per-second'='1',  " +
                " 'fields.id.kind'='sequence', " +
                " 'fields.id.start'='1'," +
                " 'fields.id.end'='1000'," +
                " 'fields.tag.length'='1'," +
                " 'fields.money.min'='100'," +
                " 'fields.money.max'='10000')");


        //使用窗口TVF实现累计窗口
        Table sqlQuery = tableEnv.sqlQuery("SELECT UNIX_TIMESTAMP(CAST(window_start AS STRING)) * 1000 ,window_end , COUNT(DISTINCT id ) ,SUM(money) as money FROM TABLE(CUMULATE(TABLE tag_money_table ,DESCRIPTOR(ts) ,INTERVAL '5' SECOND ,INTERVAL '30' SECOND  )) GROUP BY window_start ,window_end ");
        //创建Logger对象
        Logger logger = LoggerFactory.getLogger(tableDemo3.class);
        //将表转换成流
        tableEnv.toDataStream(sqlQuery,Row.class).flatMap(new FlatMapFunction<Row, String>() {
            @Override
            public void flatMap(Row value, Collector<String> out) throws Exception {
               long l = Long.parseLong(String.valueOf(value.getField("money")));
                 if ( l >= 10000L){
                    logger.info("......告警.....");
                    logger.info("-------");
                 }

            out.collect(value.toString());
            }
        }).print();

        env.execute();
    }
}

要配置log4j.properties,把这个文件夹放在src/main/resources下

### 设置###
log4j.rootLogger = debug,stdout,D,E

### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

### 输出DEBUG 级别以上的日志到=/home/hao/Desktop/error.log ###
log4j.appender.D = org.apache.log4j.DailyRollingFileAppender
log4j.appender.D.File = /home/hao/Desktop/error.log
log4j.appender.D.Append = true
log4j.appender.D.Threshold = DEBUG 
log4j.appender.D.layout = org.apache.log4j.PatternLayout
log4j.appender.D.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %m%n

### 输出ERROR 级别以上的日志到=/home/hao/Desktop/error.log ###
log4j.appender.E = org.apache.log4j.DailyRollingFileAppender
log4j.appender.E.File =/home/hao/Desktop/error.log
log4j.appender.E.Append = true
log4j.appender.E.Threshold = ERROR 
log4j.appender.E.layout = org.apache.log4j.PatternLayout
log4j.appender.E.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %m%n

日期.时间类型

DATE:由年-月-日组成的不带时区含义的日期函数类型,取值范围[0000-01-01 , 9999-12-31]
TIME,TIME(p):不带时区的时间数据类型，由 hour:minute:second[.fractional] 组成，精度达到纳秒，范围从 00:00:00.000000000 到 23:59:59.999999999。与 SQL 标准相比，不支持 leap seconds（23:59:60 和 23:59:61），语义上更接近于 java.time.LocalTime。没有提供带有时区的时间。其中 p 是秒的小数部分的位数（精度）。p 的值必须介于 0 和 9 之间（含边界值）。如果未指定精度，则 p 等于 0
TIMESTAMP,TIMESTAMP(p):不带时区的时间戳数据类型，由 year-month-day hour:minute:second[.fractional] 组成，精度达到纳秒，范围从 0000-01-01 00:00:00.000000000 到 9999-12-31 23:59:59.999999999。与 SQL 标准相比，不支持 leap seconds（23:59:60 和 23:59:61），语义上更接近于 java.time.LocalDateTime其中 p 是秒的小数部分的位数（精度）。p 的值必须介于 0 和 9 之间（含边界值）。如果未指定精度，则 p 等于 6。TIMESTAMP(p) WITHOUT TIME ZONE 等价于此类型
TIMESTAMP WITH TIME ZONE,TIMESTAMP(p) WITH TIME ZONE:带有时区的时间戳数据类型，由 year-month-day hour:minute:second[.fractional] zone 组成，精度达到纳秒，范围从 0000-01-01 00:00:00.000000000 +14:59 到 9999-12-31 23:59:59.999999999 -14:59 其中 p 是秒的小数部分的位数（精度）。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
INTERVAL YEAR TO MONTH:一组由 Year-Month Interval 组成的数据类型，其范围从 -9999-11 到 +9999-11，可以表达：

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

其中 p 是年数（年精度）的位数。p 的值必须介于 1 和 4 之间（含边界值）。如果未指定年精度，p 则等于 2。

时间间隔字面量	说明
INTERVAL '3' YEAR	时间间隔为3年
INTERVAL '3' MONTH	时间间隔为3个月
INTERVAL '3-4' YEAR TO MONTH	时间间隔为3年4个月

INTERVAL DAY TO SECOND: 一组由 Day-Time Interval 组成的数据类型。时间间隔由 +days hours:minutes:seconds.fractional 组成，其范围从 -999999 23:59:59.999999999 到 +999999 23:59:59.999999999，可以表达：

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。

时间间隔字面量	说明
INTERVAL '3' DAY	时间间隔为3天
INTERVAL '2' HOUR	时间间隔为2小时
INTERVAL '25' MINUTE	时间间隔为25分钟
INTERVAL '45' SECOND	时间间隔为45秒
INTERVAL '3 02' DAY TO HOUR	时间间隔为3天零2小时
INTERVAL '3 02:25' DAY TO MINUTE	时间间隔为3天零2小时25分
INTERVAL '3 02:25:45' DAY TO SECOND	时间间隔为3天零2小时25分45秒
INTERVAL '02:25' HOUR TO MINUTE	时间间隔为2小时25分
INTERVAL '02:25:45' HOUR TO SECOND	时间间隔为2小时25分45秒
INTERVAL '25:45' MINUTE TO SECOND	时间间隔为25分45秒

3. 动态输出表转化为输出数据

数据库中表的DML行为有3种： INSERT、UPDATE、DELETE。动态表作为类似于数据库的表的概念，也支持这3种DML操作，但是有些不同

在Flink中，将动态表分为3种类型：
1）只有更新行为，只有一行或多行但被持续更新的表。
2）只有插入行为，没有UPDATE、DELETE更改的只插入表。
3）既有插入行为也有更新行为的表。

当将动态表转化为流或将其写入外部系统时，对动态表的更改（修改）需要被转换为流上的行为，这3种类型的动态表，对应于不同类型的数据流。Flink的Table API & SQL支持3种方式的动态表上的更改（修改）。Flink的Table API & SQL支持3种方式的动态表上的更改（修改）。

（1）. Append流
Append流只支持追加写入行为，即只支持INSERT行为的动态表，不支持Update、Delete等改变已存在数据的行为。
（2）. Retract流
Retract流包含两种类型消息：add消息和retract消息。动态表的更改行为对应的消息类型如下。
1）INSERT更改转换为流上的add消息。
2）DELETE更改转换为流上的retract消息。
3）UPDATE更改转换为两条消息，即对旧记录的retract消息和新记录的add消息。

动态表转换为Retract流

（3）. Upsert流
Upsert流包含两种类型消息：update消息和delete消息。动态表转化为Upsert流必须有主键（可以是复合主键），具有主键的动态表的更改行为对应的消息类型如下。
1）INSERT、UPDATE转换为UPSERT消息。
2）DELETE转换为delete消息。

动态表到Upsert流的转换

4. SQL中指定时间属性的两种方式

如果要满⾜ Flink SQL 时间窗⼝类的聚合操作，SQL 或 Table API 中的数据源表就需要提供时间属性（相当于我们把这个时间属性在数据源表上⾯进⾏声明），以及⽀持时间相关的操作。FlinkSQL为我们提供的两种指定时间戳的方式

CREATE TABLE DDL 创建表的时候指定
可以在DataStream 中指定,在后续的DataStream转Table中使用

Event Time

CREATE TABLE user_actions (
  user_name STRING,
  data STRING,
  user_action_time TIMESTAMP(3),
  -- declare user_action_time as event time attribute and use 5 seconds delayed watermark strategy
  WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
) WITH (
  ...
);

SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);


CREATE TABLE user_actions (
  user_name STRING,
  data STRING,
  ts BIGINT,
  time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
  -- declare time_ltz as event time attribute and use 5 seconds delayed watermark strategy
  WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND
) WITH (
  ...
);

SELECT TUMBLE_START(time_ltz, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(time_ltz, INTERVAL '10' MINUTE);

// Option 1:

// extract timestamp and assign watermarks based on knowledge of the stream
DataStream<Tuple2<String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);

// declare an additional logical field as an event time attribute
Table table = tEnv.fromDataStream(stream, $("user_name"), $("data"), $("user_action_time").rowtime());


// Option 2:

// extract timestamp from first field, and assign watermarks based on knowledge of the stream
DataStream<Tuple3<Long, String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);

// the first field has been used for timestamp extraction, and is no longer necessary
// replace first field with a logical event time attribute
Table table = tEnv.fromDataStream(stream, $("user_action_time").rowtime(), $("user_name"), $("data"));

// Usage:

WindowedTable windowedTable = table.window(Tumble
       .over(lit(10).minutes())
       .on($("user_action_time"))
       .as("userActionWindow"));

Processing Time

CREATE TABLE user_actions (
  user_name STRING,
  data STRING,
  user_action_time AS PROCTIME() -- declare an additional field as a processing time attribute
) WITH (
  ...
);

SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);

DataStream<Tuple2<String, String>> stream = ...;

// declare an additional logical field as a processing time attribute
Table table = tEnv.fromDataStream(stream, $("user_name"), $("data"), $("user_action_time").proctime());

WindowedTable windowedTable = table.window(
        Tumble.over(lit(10).minutes())
            .on($("user_action_time"))
            .as("userActionWindow"));

Flink Table & SQL 时间属性
 Time Attributes | Apache Flink

2022-06-21-Flink-49(一. SQL手册)

1. 动态表

2. ⼀个 SQL\Table API 任务的代码结构

SQL上下文:TableEnvironment

SQL中的表概念

一个SQL查询案例

SQL 与 DataStream API 的转换

日期.时间类型

3. 动态输出表转化为输出数据

4. SQL中指定时间属性的两种方式

Event Time

Processing Time

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

数据库练习题