Flink SQL Table 我们一起去看 2018 中超联赛 - Flink 牛刀小试
写在前面的话
Flink 是一个新型的流式处理引擎, 作者自身只是对 Spark 底层较为熟悉, 有兴趣可以查阅我的 Spark core ,Spark Streaming 以及 Spark SQL 源码解读系列. 在这里我们只是品味一下号称第四代大数据处理引擎的 Flink, 作者也并没有深入到 Flink 底层源码级别. 请见谅如果您已经是 FLink 大牛了! 看一下 2018 中超联赛积分榜:
1 SQL Table(牛刀小试)
Table API 是以 表 为中心的声明式 DSL, 其中表可能会动态变化 (在表达流数据时).Table API 遵循(扩展的) 关系模型: 表有二维数据结构 (schema)(类似于关系数据库中的表), 同时 API 提供可比较的操作, 例如 select,project,join,group-by,aggregate 等. Table API 程序声明式地定义了 什么逻辑操作应该执行 而不是准确地确定 这些操作代码的看上去如何 . 尽管 Table API 可以通过多种类型的用户自定义函数(UDF) 进行扩展, 其仍不如 核心 API 更具表达能力, 但是使用起来却更加简洁(代码量更少). 除此之外, Table API 程序在执行之前会经过内置优化器进行优化.
你可以在表与 DataStream/DataSet 之间无缝切换, 以允许程序将 Table API 与 DataStream 以及 DataSet 混合使用.
Flink 提供的最高层级的抽象是 SQL . 这一层抽象在语法与表达能力上与 Table API 类似, 但是是以 SQL 查询表达式的形式表现程序. SQL 抽象与 Table API 交互密切, 同时 SQL 查询可以直接在 Table API 定义的表上执行.
Apache Flink 对 SQL 的支持可以追溯到一年前发布的 0.9.0-milestone1 版本. 此版本通过引入 Table API 来提供类似于 SQL 查询的功能, 此功能可以操作分布式的数据集, 并且可以自由地和 Flink 其他 API 进行组合. Tables 在发布之初就支持静态的以及流式数据(也就是提供了 DataSet 和 DataStream 相关 APIs). 我们可以将 DataSet 或 DataStream 转成 Table; 同时也可以将 Table 转换成 DataSet 或 DataStream.
The highest level abstraction offered by Flink is SQL. This abstraction is similar to the Table API both in semantics and expressiveness, but represents programs as SQL query expressions. The SQL abstraction closely interacts with the Table API, and SQL queries can be executed over tables defined in the Table API.
用户可以通过 TableEnvironment 类中的 sqlQuery() 方法执行 SQL 查询, 查询结果会以 Table 形式返回. 用户可将 Table 用于后续的 SQL 及 Table 查询, 或将其转换成 DataSet 或 DataStream, 亦可将它写入到某个 TableSink 中. 无论是通过 SQL 还是 Table API 提交的查询都可以进行无缝衔接, 系统内部会对它们进行整体优化, 并最终转换成一个 Flink 程序执行.
为了在 SQL 查询中使用某个 Table, 用户必须先在 TableEnvironment 中对其进行注册. Table 的注册来源可以是某个 TableSource, 某个现有的 Table, 或某个 DataStream 或 DataSet. 此外, 用户还可以通过在 TableEnvironment 中注册外部 Catalog 的方式来指定数据源位置.
为方便使用, 在执行 Table.toString() 方法时, 系统会自动以一个唯一名称在当前 TableEnvironment 中注册该 Table 并返回该唯一名称. 因此, 在以下示例中, Table 对象都可以直接以内联 (字符串拼接) 方式出现在 SQL 语句中.
注意: 现阶段 Flink 对于 SQL 的支持还并不完善. 如果在查询中使用了系统尚不支持的功能, 会引发 TableException .
2 上代码分析(球队粒度进行进球聚合排序)
1 进行 pojo 对象的数据封装.
2 BatchTableEnvironment tableEnv 环境生成: BatchTableEnvironment.getTableEnvironment(env);
3 Table 表生成: Table topScore = tableEnv.fromDataSet(topInput)
4 Table 表注册: tableEnv.registerTable("topScore",topScore);
5 Table 表查询: tableEnv.sqlQuery
6 Table 表转换回 DataSet: tableEnv.toDataSet
2.1 详情请参考代码
- import org.apache.flink.API.common.functions.MapFunction;
- import org.apache.flink.API.java.DataSet;
- import org.apache.flink.API.java.ExecutionEnvironment;
- import org.apache.flink.API.java.tuple.Tuple2;
- import org.apache.flink.streaming.API.datastream.DataStreamSource;
- import org.apache.flink.table.API.Table;
- import org.apache.flink.table.API.java.BatchTableEnvironment;
- public class TableSQL {
- public static void main(String[] args) throws Exception{
- ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- BatchTableEnvironment tableEnv = BatchTableEnvironment.getTableEnvironment(env);
- DataSet<String> input = env.readTextFile("C:\\CoreForBigData\\FLINK\\TopCore.csv");
- input.print();
- DataSet<TopScorers> topInput = input.map(new MapFunction<String, TopScorers>() {
- @Override
- public TopScorers map(String s) throws Exception {
- String[] splits = s.split("\t");
- return new TopScorers(Integer.valueOf(splits[0]),splits[1],splits[2],Integer.valueOf(splits[3]),Integer.valueOf(splits[4]),Integer.valueOf(splits[5]),Integer.valueOf(splits[6]),Integer.valueOf(splits[7]),Integer.valueOf(splits[8]),Integer.valueOf(splits[9]),Integer.valueOf(splits[10]));
- }
- });
- // 将 DataSet 转换为 Table
- Table topScore = tableEnv.fromDataSet(topInput);
- // 将 topScore 注册为一个表
- tableEnv.registerTable("topScore",topScore);
- Table tapiResult = tableEnv.scan("topScore").select("club");
- tapiResult.printSchema();
- Table groupedByCountry = tableEnv.sqlQuery("select club, sum(jinqiu) as sum_score from topScore group by club order by sum_score desc");
- // 转换回 dataset
- DataSet<Result> result = tableEnv.toDataSet(groupedByCountry, Result.class);
- // 将 dataset map 成 tuple 输出
- result.map(new MapFunction<Result, Tuple2<String,Integer>>() {
- @Override
- public Tuple2<String, Integer> map(Result result) throws Exception {
- String country = result.club;
- int sum_total_score = result.sum_score;
- return Tuple2.of(country,sum_total_score);
- }
- }).print();
- }
- /**
- * 源数据的映射类
- */
- public static class TopScorers {
- /**
- * 排名, 球员, 球队, 出场, 进球, 射正, 任意球, 犯规, 黄牌, 红牌
- */
- public Integer rank;
- public String player;
- public String club;
- public Integer chuchang;
- public Integer jinqiu;
- public Integer zhugong;
- public Integer shezheng;
- public Integer renyiqiu;
- public Integer fangui;
- public Integer huangpai;
- public Integer hongpai;
- public TopScorers() {
- super();
- }
- public TopScorers(Integer rank, String player, String club, Integer chuchang, Integer jinqiu, Integer zhugong, Integer shezheng, Integer renyiqiu, Integer fangui, Integer huangpai, Integer hongpai) {
- this.rank = rank;
- this.player = player;
- this.club = club;
- this.chuchang = chuchang;
- this.jinqiu = jinqiu;
- this.zhugong = zhugong;
- this.shezheng = shezheng;
- this.renyiqiu = renyiqiu;
- this.fangui = fangui;
- this.huangpai = huangpai;
- this.hongpai = hongpai;
- }
- }
- /**
- * 统计结果对应的类
- */
- public static class Result {
- public String club;
- public Integer sum_score;
- public Result() {}
- }
- }
2.2 结果展示(2018 恒大队很厉害, 进球 55 个)
3 理论升华一下
- 3.1 Create a TableEnvironment
- //***************
- // STREAMING QUERY
- //***************
- StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
- // create a TableEnvironment for streaming queries
- StreamTableEnvironment sTableEnv = TableEnvironment.getTableEnvironment(sEnv);
- //***********
- // BATCH QUERY
- //***********
- ExecutionEnvironment bEnv = ExecutionEnvironment.getExecutionEnvironment();
- // create a TableEnvironment for batch queries
- BatchTableEnvironment bTableEnv = TableEnvironment.getTableEnvironment(bEnv);
3.2 DSL 风格用法
- // get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
- StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
- // register Orders table
- // scan registered Orders table
- Table orders = tableEnv.scan("Orders");
- // compute revenue for all customers from France
- Table revenue = orders
- .filter("cCountry ==='FRANCE'")
- .groupBy("cID, cName")
- .select("cID, cName, revenue.sum AS revSum");
- // emit or convert Table
- // execute query
- 3.3 Register a DataStream or DataSet as Table
- // get StreamTableEnvironment
- // registration of a DataSet in a BatchTableEnvironment is equivalent
- StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
- DataStream<Tuple2<Long, String>> stream = ...
- // register the DataStream as Table "myTable" with fields "f0", "f1"
- tableEnv.registerDataStream("myTable", stream);
- // register the DataStream as table "myTable2" with fields "myLong", "myString"
- tableEnv.registerDataStream("myTable2", stream, "myLong, myString");
- 3.4 Convert a DataStream or DataSet into a Table
- // get StreamTableEnvironment
- // registration of a DataSet in a BatchTableEnvironment is equivalent
- StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
- DataStream<Tuple2<Long, String>> stream = ...
- // Convert the DataStream into a Table with default fields "f0", "f1"
- Table table1 = tableEnv.fromDataStream(stream);
- // Convert the DataStream into a Table with fields "myLong", "myString"
- Table table2 = tableEnv.fromDataStream(stream, "myLong, myString");
4 收工
通过 2018 中超联赛, 我们管中窥豹, 学会了 Flink SQL Table 的核心思想, 当然本文并不完善, 希望本文能够给大家带来一些收获. 辛苦成文, 彼此珍惜, 谢谢!
来源: https://juejin.im/post/5bfc018d6fb9a049d131d1a9
- 2018-11-28
- 2018-11-28