Scala编写批处理和流处理wordcount
这部分，我们在idea上使用Maven编写Scala程序实现批处理wordcount功能。
pow文件依赖配置
pow文件需要如下依赖，这里注意scala版本必须要和flink安装包适配的版本一致：
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>1.10.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.10.1</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>4.4.0</version>
<executions>
<execution>
<goals>
<goal>compile</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
批处理wordcount
之后我们准备好输入文本，并编写相应Scala代码：
hello.txt文本如下：
hello world
hello flink
hello scala
how are you
fine thank you
and you
scala文件WordCount脚本代码如下：
package com.example.wc
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
//批处理的word count
object WordCount {
def main(args: Array[String]): Unit = {
//创建一个批处理
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
//从文件中读取数据
val inputPath:String = "/Users/wenhuan/IdeaProjects/FlinkTutorial/src/main/resources/hello.txt"
val inputDataSet:DataSet[String] = env.readTextFile(inputPath)
//对数据进行转换处理统计，先分词
val resultDataSet:DataSet[(String,Int)] = inputDataSet
.flatMap(_.split(" "))
.map((_,1))
.groupBy(0)
.sum(1)
//打印输出
resultDataSet.print()
}
}
流处理wordcount
scala文件WordCount脚本代码如下：
package com.example.wc
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
// 流式处理word count
object StreamWordCount {
def main(args: Array[String]): Unit = {
//创建流处理的执行环境
val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//设置并行度
//env.setParallelism(2)
//从外部命令中提取参数，作为socket主机名和端口号
val paramTool:ParameterTool = ParameterTool.fromArgs(args)
val host:String = paramTool.get("host")
val port:Int = paramTool.getInt("port")
//接收一个socket文本流
val inputDataStream:DataStream[String] = env.socketTextStream(host,port)
//进行转化处理统计
val resultDataStream:DataStream[(String,Int)] = inputDataStream
.flatMap(x => x.split(" "))
.filter(_.nonEmpty)
.map((_,1))
.keyBy(0)
.sum(1)
resultDataStream.print()
//启动任务执行
env.execute("stream word count")
}
}
idea运行时需要配置Program arguments如下：
--host localhost --port 7777
开启本地nc -lk服务：
nc -lk 7777
然后执行上述代码即可。
任务提交执行
flink任务提交执行有两种方式，第一种是在网页直接提交，另一种是命令行方式。
命令行方式
提交任务命令如下：
./bin/flink run -c com.example.wc.StreamWordCount -p 1 ./FlinkTutorial-1.0-SNAPSHOT-jar-with-dependencies.jar --host localhost --port 7777
查看当前正在运行任务命令如下：
./bin/flink list
停止正在运行任务的命令如下:
./bin/flink cancel <jobid>
