零基础学 Flink:Flink SQL(上)

前面几篇内容,我们结合案例来介绍了,两流Join,热销榜,以及状态容错,今天我们依旧基于这个数据,来说说Flink SQL,如果对原理有兴趣的同学,也可以移步到《 Stream SQL 的执行原理与 Flink 的实现 》 ,去了解相关内容。

今天我们分几步来介绍,首先什么是动态表,如何注册,数据流如何转换。本文配图主要来自官方文档。

SQL和关系代数设计的时候,并没有考虑流计算,所以流计算和关系数据的计算,有很多概念上的差异。

首先这是一张Flink官方的表

关系代数 / SQL 流计算
关系数据可以表示成一个元组的集合。 一条流是由一条无界的元组数据流组成
一条查询时,包含完整的输入数据。

计算流数据的时候,无法得到所有数据,必须要等待有合适的数据流入。

批查询在终止时,结果是有固定大小的。

流式查询会根据接收到的记录不断更新其结果,而且永远不会完。

尽管有这些不同,但是并非使用关系计算流数据变得不可能,下面我们就来详细说说。

动态表

动态表可以说是Flink Table API 和 SQL的核心,动态表可以像普通关系型数据表一样被查询,只是他吐出的数据,是一个持续的数据流。

  1. 一个流首先被定义转化成动态表

  2. 对动态表进行持续查询,然后这个查询的结果还要被定义成动态表

  3. 最后动态表还需要重新转化成流

如何定义一个动态表?

要想像传统关系查询一样处理流数据,就需要将流转换成一个表,那么如何转换这个表呢?

这个观念大家应该很好理解,我们截取一个流的片段,然后流输入的每一条,其实就相当于关系型数据的一条记录。在关系型数据库里,我们需要首先定义数据表,而流数据在处理的时候,才可以定义元数据。

下面是我们在flink里如何将流定义成动态表,

tableEnv.registerDataStream("orders", oraderStream, "rowtime.rowtime, c1,c2, dm ,v ");

orders 是我们定义的表名, orderStream 则是一条数据流, 后面的字符串参数是流里面数据的列名,这里要注意的是 rowtime.rowtime 它是定义时间窗口的时间属性,在我们的数据列里,可以将一列指定为rowtime,也可以添加一列来辅助计算。

持续查询

在我们进行普通的数据映射和过滤的时候,流和关系表的计算几乎是没什么区别,只是有界与无界的区分。

当进行聚合的时候,数据持续输入,都会对聚合结果有影响,例如下图,对用户点击进行统计的时候,随着时间增长,用户点击的发生,其点击数据是会持续增加的,这就造成了持续查询的数据在不停的更新。

下图是有时间窗口的聚合,在时间窗口内,聚合可以当成一个小的关系型聚合计算来理解。

动态表输出流

动态表输出力流有三种模式 Append-only stream, Retract streamUpsert stream

Append-only stream 只有在动态Table仅通过INSERT更改修改时才能使用此模式,即它仅附加,并且以前发出的结果永远不会更新。

Retract stream 此模式。返回值是boolean类型。它用true或false来标记数据的插入和撤回,返回true代表数据插入,false代表数据的撤回

Upsert stream 和 Retract stream最大的区别在于,更新数据的时候只使用一条编码消息,所以效率更高。

代码案例

我们还是以几篇文章使用的订单流进行。我设计了两个查询,

  • 一个是按时间窗口和商品类型对销售额进行聚合

  • 另外一个是只按商品类型对销售额进行聚合。

前面部分消费kafka的部分没有什么变化,只是在获取初始数据流的时候,将首字段设置成了Timestamp类型。并在获取流的时候,加入watermarker。

然后就是前文提到的

tableEnv.registerDataStream("orders", oraderStream, "rowtime.rowtime, c1,c2, dm ,v ");

将流注册成动态表,并设置元数据,注意

rowtime.rowtime

这个写法,前文有详细说明。

在有时间聚合的动态表转换的时候,我使用了

toAppendStream

没有时间聚合的情况,使用了

toRetractStream

下面是完整代码:

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import java.io.IOException;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;
public class App {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Map properties= new HashMap();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("group.id", "test");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "30000");
//        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("topicOrder", "order");
        ParameterTool parameterTool = ParameterTool.fromMap(properties);
        FlinkKafkaConsumer010 consumer010Order = new FlinkKafkaConsumer010(
                parameterTool.getRequired("topicOrder"), new DeserializationSchema() {
            @Override
            public TypeInformation getProducedType() {
                return TypeInformation.of(new TypeHint<Tuple5<Timestamp,String,Integer,String,Integer>>(){});
            }

            @Override
            public Tuple5<Timestamp,String,Integer,String,Integer> deserialize(byte[] message) throws IOException {
                //%d,%s,%d,%s,%d
                String[] res = new String(message).split(",");
                Timestamp timestamp = new Timestamp(Long.valueOf(res[0]));
                String catlog = res[1];
                Integer subcat = Integer.valueOf(res[2]);
                String dm = res[3];
                Integer value = Integer.valueOf(res[4]);
                Time eventTime = new Time(System.currentTimeMillis());
                return Tuple5.of(timestamp,catlog,subcat,dm,value);
            }

            @Override
            public boolean isEndOfStream(Object nextElement) {
                return false;
            }
        }, parameterTool.getProperties());
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        DataStream<Tuple5<Timestamp,String,Integer,String,Integer>> oraderStream = env.addSource(consumer010Order).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Tuple5<Timestamp,String,Integer,String,Integer>>() {
            @Override
            public long extractAscendingTimestamp(Tuple5<Timestamp,String,Integer,String,Integer> value) {
                return value.f0.getTime();
            }
        });;
        tableEnv.registerDataStream("orders", oraderStream, "rowtime.rowtime, c1,c2, dm ,v ");
        String sql = "select sum(orders.v),CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR)),TUMBLE_END(rowtime, INTERVAL '10' SECOND)  from orders group by CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR)),TUMBLE(rowtime, INTERVAL '10' SECOND)";
        //sql = "select sum(orders.v),CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR))  from orders group by CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR))";
        Table result1 = tableEnv.sqlQuery(sql);
        result1.printSchema();
        tableEnv.toAppendStream(result1, Row.class).print();
//        tableEnv.toRetractStream(result1, Row.class).print();
        env.execute("sql ");
    }
}

结果

select

sum(orders.v),

CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR)),

TUMBLE_END(rowtime, INTERVAL '10' SECOND)

from  orders

group by

CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR)),

TUMBLE(rowtime, INTERVAL '10' SECOND)

对应的执行结果

select

sum(orders.v),

CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR))

from orders

group by

CONCAT(orders.c1,CAST(orders.c2 AS VARCHAR))

对应的执行结果

写在结尾:

Joins are a common and well-understood operation in batch data processing to connect the rows of two relations. However, the semantics of joins on dynamic tables are much less obvious or even confusing.

https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/streaming/joins.html

这段话也算是官方吐槽吧,Join两条流的时候,确实用SQL表达会有很大的障碍,目前我是打算啃啃这块骨头的,如果一两周内没更新下篇,就是翻车了,啃的时候门牙崩了,请读者们多做自我检讨吧......

参考文献:

https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/sourceSinks.html#defining-a-streamtablesource

https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/streaming/joins.html

https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/streaming/dynamic_tables.html

https://gist.github.com/mustafaakin/457859b8bf703c64029071c1139b593d

https://blog.csdn.net/aa518189/article/details/87816139

学学漫威,文尾贴个彩蛋吧,下面是来自某群里的吐槽,个人观点倒是积极的,俗话说,褒贬是买主,喝彩是闲人。 希望Flink越来越好吧....

我来评几句
登录后评论

已发表评论数()

相关站点

+订阅
热门文章