11.1.2、flink概述_Wordcount案例

WordCount案例

flink中的算子为有状态算子
—相同的key被分到同一个task中
—注意隐式转换的错误

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

object Demo01WordCount {
    

  def main(args: Array[String]): Unit = {
    
    //创建flink的环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 使用nc来模拟流数据
      * yum install nc
      * nc -lk 8888
      */
      //设置并行度,不设置的话，不同数据在不同task中，12核就是12个数字前面
      //数据被分到不同的核中计算数据
    env.setParallelism(1)
    //读取scoket的数据
    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)

    //将数据拆开
    val wordsDS: DataStream[String] = lineDS.flatMap(_.split(","))

    //将数据后面个数设置为1
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))

    //使用KeyBy进行分组，将不同的数据放在不同的reduce task中
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    //求和,对第二个元素进行求和,传入下标，按下标的数据进行分，下标从0开始
    //有状态算子
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    //打印输出
    countDS.print()

    //启动，执行流，7*24小时执行任务
    env.execute()
  }

}