Flink的两种WordCount

流处理WordCount

import org.apache.flink.streaming.api.scala._
//流处理wordcount
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    //创建流处理执行环境
    val env = StreamExecutionEnvironment
      .getExecutionEnvironment


    //接受一个socket文本流
    val inputDataStream = env.socketTextStream("localhost",777)


    //进行转换处理统计
    val resultDataStream: DataStream[(String, Int)] = inputDataStream
      .flatMap(x => {
        x.split(" ")
      })
      .filter(_.nonEmpty)   //不为空
      .map((_,1))
      .keyBy(0)
      .sum(1)


    resultDataStream.print()


    //事件驱动 起一个进程 监听端口 等数据来做计算    启动任务执行  也就是调用环境
    //需要nc 启动开启端口  创建一个端口  相当于启动一个socket服务
    //nc -lk（一直保持） prot      也就是 nc -lk 777
    env.execute("stream word count")   //当前执行的job名字
  }
}

批处理WordCount

//批处理的word count


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


    //从文件中读取数据
    val inputPath = "/Users/edy/IdeaProjects/flinksql/src/main/resources/hello.txt"
    val inputDataSet: DataSet[String] = env.readTextFile(inputPath)


    //对数据进行转换处理统计，先粉刺，再按照word、进行分组 最后进行聚合统计




    val resultDataSet = inputDataSet
      .flatMap(_.split(" "))
      .map((_,1))
      .groupBy(0)    //下标是按第一个元素进行key分组
      .sum(1)   //对所有数据的第二个元素求和


    resultDataSet.print()