本系列文章将从热词统计需求讲起，讲解flink在实际项目中的应用、部署、容错。项目中的代码为scala所编写，没有用 java 的原因是scala语言看起来更加简练，同时与java语言的兼容性较好，可以直接引用java中的类。

定义数据格式

本篇文章会讲述热词统计最核心的一个功能实现，即消息传递过来的是搜索的单词，程序中进行统计。基于此我们定义传递的数据格式为 ${timestamp},${word} ，之所以有timestamp是为了标注搜索时间，用于修正flink中的watermask

离线数据模拟

定义好了数据格式之后，我们可以将其转换为scala中的Tuple，这里我们先利用fromElements api 创建离线数据用于调试

val currentTimeStamp = System.currentTimeMillis()

val dataStream = env.fromElements(

(currentTimeStamp,"word1"),

(currentTimeStamp+1,"word2"),

(currentTimeStamp+2,"word1"),

(currentTimeStamp+3,"word3"),

)

复制代码

具体计算

我们要实现统计的功能，即求和，需要flink中的sum这个operator，通过[flink-1.8文档/操作符]发现，只有keyedStream这个对象才有sum方法，所以在执行sum之前得先调用keyBy方法将dataStream转换为keyedStream。接下来是这两个方法具体的传参，我们有两个字段，时间戳和单词，要对单词进行keyBy操作需传入1(下标从0开始)，但是sum方法要传什么呢？这个时候我们发现我们的对象中没有可以用来求和的，所以我们需要将数据结构从 ${timestamp},${word} 转换为 ${timestamp},${word},${count} ，这就得使用map方法了，其中count为1。经过如上分析，有如下代码

dataStream

.map(x => (x._1, x._2, 1))

.keyBy(1)

.sum(2)

.print()

复制代码

为便于调试，最后的结果输出为到控制台即为

6> (1556680015646,word2,1)

12> (1556680015648,word3,1)

10> (1556680015647,word1,1)

10> (1556680015647,word1,2)

输出结果中开头的数字即为线程的标示，这里我们看出，word2和word3分别是不同的线程，而word1则为同一个线程

疑问解答

为什么这里关于word1的统计有两次输出呢？ 因为这里是实时流式计算，来一条数据，各个operator就会为它计算一次，所以最后都有一次对应的输出

那什么时候输出的个数会变少呢？ 在使用window的时候，多个数据会按照一定的规则(根据个数或者根据时间段)聚集在一个window中，然后每一个window就只有一个输出结果

小技巧

这里我们使用的scala语言来编写flink项目，需要注意的是要在pom.xml文件中加上如下代码，使其能够支持scala编译

net.alchim31.maven

scala-maven-plugin

scala-compile-first

process-resources

compile

scala-test-compile

process-test-resources

testCompile

-Xms128m

-Xmx512m

复制代码

在项目的引用中，需要引用scala后缀的，如

2.11

1.8.0

org.apache.flink

flink-streaming-scala_${scala.binary.version}

${flink.version}

复制代码

文件中的引用，在你输入 StreamExecutionEnvironment 的时候，Intellij会提示你引入 org.apache.flink.streaming.api.scala. StreamExecutionEnvironment ，但是当你调用 env.fromElements() 方法的时候你会编译器会有类型错误的提示

这个时候你只需要把文件头部的引用改为

import org.apache.flink.streaming.api.scala._

就可以了