Flink学习（四） Flink Table & SQL 实现wordcount Java版本

Flink Table & SQL WordCount
Flink SQL 是 Flink 实时计算为简化计算模型，降低用户使用实时计算门槛而设计的一套符合标准 SQL 语义的开发语言。

一个完整的 Flink SQL 编写的程序包括如下三部分。

Source Operator：是对外部数据源的抽象, 目前 Apache Flink 内置了很多常用的数据源实现，比如 MySQL、Kafka 等。
Transformation Operators：算子操作主要完成比如查询、聚合操作等，目前 Flink SQL 支持了 Union、Join、Projection、Difference、Intersection 及 window 等大多数传统数据库支持的操作。
Sink Operator：是对外结果表的抽象，目前 Apache Flink 也内置了很多常用的结果表的抽象，比如 Kafka Sink 等。
我们也是通过用一个最经典的 WordCount 程序作为入门，上面已经通过 DataSet/DataStream API 开发，那么实现同样的 WordCount 功能， Flink Table & SQL 核心只需要一行代码：

//省略掉初始化环境等公共代码
SELECT word, COUNT(word) FROM table GROUP BY word;

首先，整个工程中我们 pom 中的依赖如下图所示：

<dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-java</artifactId>
         <version>1.10.0</version>
     </dependency>
<dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-streaming-java_2.11
         <version>1.10.0</version>
</dependency>
<dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-table-api-java-bridge_2.11</artifactId>
         <version>1.10.0</version>
</dependency>
<dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-table-planner-blink_2.11</artifactId>
         <version>1.10.0</version>
</dependency>
<dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-table-planner_2.11</artifactId>
         <version>1.10.0</version>
</dependency>
     <dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
         <version>1.10.0</version>
</dependency>

第一步，创建上下文环境：

ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);

第二步，读取一行模拟数据作为输入：

String words = "hello flink hello lagou";
String[] split = words.split("\\W+");
ArrayList<WC> list = new ArrayList<>();

for(String word : split){
    WC wc = new WC(word,1);
    list.add(wc);
}
DataSet<WC> input = fbEnv.fromCollection(list);

第三步，注册成表，执行 SQL，然后输出：

//DataSet 转sql, 指定字段名
Table table = fbTableEnv.fromDataSet(input, "word,frequency");
table.printSchema();

//注册为一个表
fbTableEnv.createTemporaryView("WordCount", table);

Table table02 = fbTableEnv.sqlQuery("select word as word, sum(frequency) as frequency from WordCount GROUP BY word");

//将表转换DataSet
DataSet<WC> ds3  = fbTableEnv.toDataSet(table02, WC.class);
ds3.printToErr();

整体代码结构如下：

package wyh.tableApi;import org.apache.flink.api.java.DataSet;import org.apache.flink.api.java.ExecutionEnvironment;import org.apache.flink.api.java.operators.DataSource;import org.apache.flink.table.api.Table;import org.apache.flink.table.api.java.BatchTableEnvironment;import java.util.ArrayList;public class WCTableApi {    public static void main(String[] args) {        ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();        BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);        String words="hello flink hello shujia";        String[] split = words.split("\\W+");        ArrayList<WC> list = new ArrayList<>();        for (String word : split) {            WC wc = new WC(word, 1L);            list.add(wc);        }//        DataSet<WC> input = fbEnv.fromCollection(list);        DataSource<WC> input = fbEnv.fromCollection(list);        Table table = fbTableEnv.fromDataSet(input, "word,frequency");        table.printSchema();        fbTableEnv.createTemporaryView("wordcount",table);        Table table1 = fbTableEnv.sqlQuery("select word,sum(frequency) as frequency from wordcount group by word");        DataSet<WC> ds3 = fbTableEnv.toDataSet(table1, WC.class);        try {            ds3.printToErr();        } catch (Exception e) {            e.printStackTrace();        }    }    public static class WC{        public String word;        public Long frequency;        public WC() {        }        public WC(String word, Long frequency) {            this.word = word;            this.frequency = frequency;        }        @Override        public String toString() {            return "WC{" +                    "word=‘" + word + ‘\‘‘ +                    ", frequency=" + frequency +                    ‘}‘;        }    }}
 

我们直接运行该程序，在控制台可以看到输出结果：