Scala Flink之WordCount的两种处理案列

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


object WordCount {
  def main(args: Array[String]): Unit = {
    // 创建一个批处理的执行环境
    val evn = ExecutionEnvironment.getExecutionEnvironment

    // 读取数据
    val inputPath = "E:\Program Files\scala_flink\src\main\resources\WordCount.txt"
    val inpatDataSet = evn.readTextFile(inputPath)

    // 对数据进行转换，先分词，再按照word进行分组，最后进行聚合统计
    val resultDataSet = inpatDataSet
      .flatMap(_.split(" "))
      .map((_,1))
      .groupBy(0) // 以第一个元素作为key，进行分组
      .sum(1) // 对所有数据的第二个元素求和

    resultDataSet.print()
  }
}

测试方式：创建一个txt文件，输入测试数据
执行结果:

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object StreamWordCount {
  def main(args: Array[String]): Unit = {
    // 创建一个流处理的执行环境
    val evn = StreamExecutionEnvironment.getExecutionEnvironment
    
    // 全局配置并行度
    evn.setParallelism(8)

    // 从外部命令中提取参数，作为socket主机名和端口号
    val paramTool:ParameterTool = ParameterTool.fromArgs(args)
    val host:String = paramTool.get("host")
    val port:Int = paramTool.getInt("port")
    val inputDataStream = evn.socketTextStream(host,port)

    // 进行转换处理操作
    val resultDataSteam = inputDataStream
      .flatMap(_.split(" "))
      .filter(_.nonEmpty)
      .map((_,1))
      .keyBy(0)
      .sum(1)

    resultDataSteam.print()

    // 启动执行任务
    evn.execute()
  }
}

测试方式：
1、执行前先配置执行信息


2、打开虚拟机后执行代码，然后打开虚拟机命令输入看板

测试结果：