- 新建maven工程
- 添加flink相关依赖到pom.xml中
4.0.0
com.suben.bigdata
flink-basic
1.0-SNAPSHOT
8
8
org.apache.flink
flink-scala_2.12
1.10.1
org.apache.flink
flink-streaming-scala_2.12
1.10.1
org.slf4j
slf4j-simple
1.7.25
compile
net.alchim31.maven
scala-maven-plugin
3.4.6
compile
org.apache.maven.plugins
maven-assembly-plugin
3.0.0
jar-with-dependencies
make-assembly
package
single
- 添加scala框架支持
- 准备批处理测试数据，新建wc.txt，添加如下内容：
I love Guizhou
I love my home
I love Flink
I love Bigdata
- 编写批处理的wordcount，新建WordCountBatch的Object，代码如下：
import org.apache.flink.api.scala.{AggregateDataSet, DataSet, ExecutionEnvironment, createTypeInformation}
object WordCountBatch {
def main(args: Array[String]): Unit = {
// 1. 初始化环境
val env = ExecutionEnvironment.getExecutionEnvironment
val lineDS: DataSet[String] = env.readTextFile("E:\IdeaProjects\bigdata-sets002\flink-basic\data\wc.txt")
// 2. 读取数据并进行转换
val aggregateDS: AggregateDataSet[(String, Int)] = lineDS.flatMap(_.split(" "))
.map(x => (x, 1))
.groupBy(0)
.sum(1)
// 3. 打印结果
aggregateDS.print()
}
}
- 运行结果如下：
- 安装nc，我用的是Windows版本的Netcat，下载后解压，启动即可，如下图所示：
然后在控制台输入如下文字即可：I love Guizhou
I love my home
I love Flink
I love Bigdata
- 编写流处理的wordcount，新建WordCountStream的Object类，然后添加如下代码即可：
import org.apache.flink.streaming.api.scala._
object WordCountStream {
def main(args: Array[String]): Unit = {
// 1. 初始化环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 读取数据并进行转换
// 接收一个socket文本流
val inputDataStream: DataStream[String] = env.socketTextStream("localhost",6666)
// 进行转化处理统计
val resultDataStream: DataStream[(String, Int)] = inputDataStream
.flatMap(_.split(" "))
.filter(_.nonEmpty)
.map((_, 1))
.keyBy(0)
.sum(1)
// 3. 为结果好看些，设置并发度为1
resultDataStream.print().setParallelism(1)
// 启动任务执行
env.execute("stream word count")
}
}
- 运行代码（运行前确认下nc需要处于启动状态），正常的结果如下图所示：
