Flink基础篇：数据界的HelloWorld和Flink SQL

实时计算在未来是每个数据人应该掌握的一项技能。随着5G技术的普及，大量可穿戴设备和终端的面世，物联网技术及应用，会产生海量的，数据价值密度不等的数据。那数据处理、计算、反馈的实时性要求会越来越迫切。要真正的了解和掌握那肯定要实践，今天我们将从一个数据界最经典的Hello World程序，WordCount 作为切入点，并且同时使用 SQL 方式进行实现，为以后的持续探索打好基础。

首先从环境搭建入手，介绍如何搭建本地调试环境的脚手架工具；
然后分别从DataSet（批处理）和 DataStream（流处理）两种方式如何进行单词计数；
最后是Flink Table 和 SQL 的使用。

Flink 开发环境

通常任何计算框架在实际生产环境中都是以集群的形式运行，而我们调试代码大多数会在本地搭建一个模板工程。

Flink 一个以 Java 及 Scala 作为开发语言的开源大数据项目，通常我们推荐使用 Java 来作为开发语言，Maven 作为编译和包管理工具进行项目构建和编译。对于大多数开发者而言，JDK、Maven 和 Git 这三个开发工具是必不可少的。

关于 JDK、Maven 和 Git 的安装建议如下表所示：

工具	版本建议	备注
JDK	1.8或以上
Maven	建议Maven3.5	https://maven.apache.org/download.cgi
Git	无要求，较新版本即可	Git下载：https://git-scm.com/downloads flink仓库：https://github.com/apache/flink
系统	建议Linux、Mac，最方便。
IDE	推荐Idea

工程创建

我们可以通过 IDE 创建工程，添加 Maven 依赖，或者直接用 mvn 命令创建应用：

mvn archetype:generate  \
    -DarchetypeGroupId=org.apache.flink \
    -DarchetypeArtifactId=flink-quickstart-java \
    -DarchetypeVersion=1.10.0

通过指定 Maven 工程的三要素，即 GroupId、ArtifactId、Version 来创建一个新的工程。

同时 Flink 给我们提供了更为方便的创建 Flink 工程的方法：

curl https://flink.apache.org/q/quickstart.sh | bash -s 1.10.0

我们在终端直接执行该命令：

直到出现 Build Success 信息，我们可以在本地目录看到一个已经生成好的名为 quickstart 的工程。

这里需要的注意的是，自动生成的项目 pom.xml 文件中对于 Flink 的依赖注释掉 scope：

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-java</artifactId>
   <version>${flink.version}</version>
   <!--<scope>provided</scope>-->
</dependency>
<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
   <version>${flink.version}</version>
   <!--<scope>provided</scope>-->
</dependency>

DataSet WordCount

WordCount 程序是大数据领域的入门HelloWorld程序，俗称“单词计数”。用来统计一段文字每个单词的出现次数，该程序主要分为两个部分：一部分是将文字拆分成单词；另一部分是单词进行分组计数并打印输出结果。

整体代码实现如下：

    public static void main(String[] args) throws Exception {

        // 创建Flink任务运行的环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 创建DataSet，数据是一行一行的文本
        DataSet<String> text = env.fromElements(
                "Flink Hello World",
                "Flink Flink World",
                "Hive Elastic Search",
                "Hadoop Hive zookeeper",
                "Hello World Hello"
        );
        // 通过Flink内置的转换函数进行计算
        DataSet<Tuple2<String, Integer>> counts =
                text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception {
                        // 将文本分割
                        String[] tokens = value.toLowerCase().split("\\W+");

                        for (String token : tokens) {
                            if (token.length() > 0) {
                                collector.collect(new Tuple2<String, Integer>(token, 1));
                            }
                        }
                    }
                })
                        .groupBy(0)
                        .sum(1);
        //结果打印
        counts.printToErr();
    }

上面是实现WordCount最简洁的一种，整个过程有以下几个步。

首先，我们需要创建 Flink 的上下文运行环境：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

然后，使用 fromElements 函数创建一个 DataSet 对象；

接着，使用该DataSet 对象包含了我们的数据，使用 FlatMap、GroupBy、SUM 函数进行转换。

最后，直接在控制台打印输出。

我们可以直接右键运行一下 main 方法，在控制台会出现我们打印的计算结果：

image-20200523193730083.png

DataStream WordCount

为了模仿一个流式计算环境，我们选择监听一个本地的 Socket 端口，并且使用 Flink 中的滚动窗口，每 5 秒打印一次计算结果。代码如下：

public class StreamingJob {

    public static void main(String[] args) throws Exception {

        // 创建流式任务执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 监听本地9000端口
        DataStream<String> text = env.socketTextStream("127.0.0.1", 9000, "\n");

        // 将接受的数据进行分组、窗口计算、聚合输出
        DataStream<WordWithCount> windowCounts = text
                .flatMap(new FlatMapFunction<String, WordWithCount>() {
                    @Override
                    public void flatMap(String value, Collector<WordWithCount> out) {
                        for (String word : value.split("\\s")) {
                            out.collect(new WordWithCount(word, 1L));
                        }
                    }
                })
                .keyBy("word")
                .timeWindow(Time.seconds(5), Time.seconds(1))
                .reduce(new ReduceFunction<WordWithCount>() {
                    @Override
                    public WordWithCount reduce(WordWithCount a, WordWithCount b) {
                        return new WordWithCount(a.word, a.count + b.count);
                    }
                });

        // 用并行度为1的单线程打印结果
        windowCounts.print().setParallelism(1);

        env.execute("Socket Window WordCount");
    }

    // Data type for words with count
    public static class WordWithCount {

        public String word;
        public long count;

        public WordWithCount() {}

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + " : " + count;
        }
    }
}

整个流式计算的过程分为以下几步。

首先创建一个流式计算环境：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

然后进行监听本地 9000 端口，将接收的数据进行拆分、分组、窗口计算并且进行聚合输出。代码中使用了 Flink 的窗口函数。

我们在本地使用 netcat 命令启动一个端口：

nc -lk 9000

然后直接运行我们的 main 方法：

INFO  org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction  - Connecting to server socket 127.0.0.1:9000

可以看到，工程启动后开始监听 127.0.0.1 的 9000 端口。

在 nc 中输入：

$ nc -lk 9000
hello world flink
hello flink flink

可以在控制台看到：

world : 1
hello : 2
flink : 3

Flink Table & SQL WordCount

Flink SQL是为了简化计算模型、降低用户使用实时计算门槛而设计的一套符合标准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 程序作为FlinkSQL入门,上面已经通过 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 env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment batchTableEnv = BatchTableEnvironment.create(env);

读取一行模拟数据作为输入：

String words = "hello flink hello world hello es hello stream flink";
String[] split = words.split("\\W+");
ArrayList<WordCount> list = new ArrayList<>();
for(String word : split){
    WordCount wc = new WordCount(word, 1);
    list.add(wc);
}
DataSet<WordCount> input = env.fromCollection(list);

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

//DataSet->Table, 指定字段名
Table table = batchTableEnv.fromDataSet(input, "word,frequency");
table.printSchema();
//注册为一个表
batchTableEnv.createTemporaryView("WordCount", table);

Table table02 = batchTableEnv.sqlQuery("select word as word, sum(frequency) as frequency from WordCount GROUP BY word");
//转Table->DataSet
DataSet<WordCount> ds  = batchTableEnv.toDataSet(table02, WordCount.class);
ds.printToErr();

整体代码结构如下：

public class WordCountSQL {

    public static void main(String[] args) throws Exception{
        //获取运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //创建一个tableEnvironment
        BatchTableEnvironment batchTableEnv = BatchTableEnvironment.create(env);

        String words = "hello flink hello world hello es hello stream flink";

        String[] split = words.split("\\W+");
        ArrayList<WordCount> list = new ArrayList<>();

        for(String word : split){
            WordCount wc = new WordCount(word, 1);
            list.add(wc);
        }

        DataSet<WordCount> input = env.fromCollection(list);
        //DataSet转Table, 指定字段名
        Table table = batchTableEnv.fromDataSet(input, "word,frequency");
        table.printSchema();
        //注册为一个表
        batchTableEnv.createTemporaryView("WordCount", table);

        Table table02 = batchTableEnv.sqlQuery("select word as word, sum(frequency) as frequency from WordCount GROUP BY word");
        //Table转DataSet
        DataSet<WordCount> ds  = batchTableEnv.toDataSet(table02, WordCount.class);
        ds.printToErr();
    }

    public static class WordCount {
        public String word;
        public long frequency;

        public WordCount() {}

        public WordCount(String word, long frequency) {
            this.word = word;
            this.frequency = frequency;
        }

        @Override
        public String toString() {
            return  word + ", " + frequency;
        }
    }
}

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

flink, 2
world, 1
hello, 4
es, 1
stream, 1

总结

本篇我们介绍了 Flink 的工程创建，如何搭建调试环境，同时以WordCount 单词计数这一经典的例子场景用 Flink 进行了实现，和初步体验了 Flink SQL 的强大和易用的特点。
关注我