批处理WordCount
第一步:创建批处理执行环境
val env:ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
然后这里需要import:
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.java.ExecutionEnvironment
注意:使用scala api又需要引入java东西,所有很多类名称一样,所以引入的时候要注意引入的名称。这里引入scala下的执行环境。
第二步:读取输入文件
resources文件夹下创建一文件以便于输入
val inputPath:String = "F:\\idea1\\scalatest113\\src\\main\\resources\\hello.txt"
val inputDataSet:DataSet[String] = env.readTextFile(inputPath)
第三步:对数据进行转换处理
//同时需要引入隐式转换
import org.apache.flink.api.scala._
val resultDataSet:DataSet[(String,Int)] = inputDataSet
.flatMap(_.split(" ")) // 分词
.map((_,1))
.groupBy(0) //按照map第一个元素分组(下标索引取值)
.sum(1) // 对map第二个元素进行求和(下标索引取值)
由于需要隐式转换所以需要导入
import org.apache.flink.api.scala._
对比于执行环境需要的导入
import org.apache.flink.api.scala.ExecutionEnvironment
发现执行环境已经被包含需要,可以省略
import org.apache.flink.api.scala.ExecutionEnvironment
第四步:控制台打印输出结果
resultDataSet.print()
输出结果展示
(scala,1)
(you,3)
(flink,1)
(world,1)
(hello,3)
(and,1)
(are,1)
(thank,1)
(fine,1)
(how,1)
完整代码
package com.erke.wc
//引入隐式转换和执行环境
import org.apache.flink.api.scala._
object wordcount {
def main(args: Array[String]): Unit = {
// 创建批处理执行环境
val env:ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
// 读取批处理文件
val inputPath:String = "F:\\idea1\\scalatest113\\src\\main\\resources\\hello.txt"
val inputDataSet:DataSet[String] = env.readTextFile(inputPath)
// 对文件进行解析转换
val resultDataSet:DataSet[(String,Int)] = inputDataSet
.flatMap(_.split(" ")) // 分词
.map((_,1)) // 转换 hello --> (hello,1)
.groupBy(0) // 按下标索引进行分组
.sum(1) // 按照下标索引进行求和
resultDataSet.print()
}
}