sum

在对datastream keyby后使用sum函数聚合

package com.stanley.wordcount

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

/**
  * Created by admin on 2020/7/2.
  */
object SumWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //并行度设置为1
    env.setParallelism(1)
    //读取文本流数据
    val inputDataStream:DataStream[String] = env.socketTextStream("node1",9999)
    val outputDataStream:DataStream[(String,Int)] = inputDataStream.flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(0)
      .sum(1)
    outputDataStream.print("sum_wordcount")
    env.execute("wc test")
  }
}

processfunction

调用最底层processfunction，将count保存成一个keyedstate

package com.stanley.wordcount

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector
/**
  * Created by admin on 2020/7/2.
  */
object ProcessWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val inputDataStream:DataStream[String] = env.socketTextStream("node1",9999)
    val outputDataStraem:DataStream[(String,Int)] = inputDataStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(0)
      //调用新建的MyProcessFunction
      .process(new MyProcessFunction)
    outputDataStraem.print("process_wordcount")
    env.execute("wc test")
  }
}

class MyProcessFunction extends KeyedProcessFunction[Tuple,(String,Int),(String,Int)]{
  //创建一个countState
  private var countState:ValueState[Int] = _
  override def open(parameters: Configuration): Unit = {
    //初始化countState
    countState = getRuntimeContext.getState[Int](new ValueStateDescriptor[Int]("count",classOf[Int]))
  }

  override def processElement(i: (String, Int), context: KeyedProcessFunction[Tuple, (String, Int), (String, Int)]#Context, collector: Collector[(String, Int)]): Unit = {
    //取出count
    var count = countState.value()
    count+=1
    //更新countState
    countState.update(count)
    collector.collect((i._1,count))
  }
}

RichMapFunction

RichMapFunction和ProcessFunction一样都是实现了AbstractRichFunction，所以同样拥有生命周期方法和运行时上下文，以及keyed state

package com.stanley.wordcount

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

/**
  * Created by admin on 2020/7/2.
  */
object RichWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val inputDataStream:DataStream[String] = env.socketTextStream("node1",9999)
    val outputDataStraem:DataStream[(String,Int)] = inputDataStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(0)
      //调用MyRichMapFunction
      .map(new MyRichMapFunction)
    outputDataStraem.print("rich_wordcount")
    env.execute("wc test")
  }
}

class MyRichMapFunction extends RichMapFunction[(String,Int),(String,Int)]{
  private var countState:ValueState[Int] = _

  override def open(parameters: Configuration): Unit = {
    countState = getRuntimeContext.getState[Int](new ValueStateDescriptor[Int]("count",classOf[Int]))
  }

  override def map(in: (String,Int)): (String, Int) = {
    var count = countState.value()
    count+=1
    countState.update(count)
    (in._1,count)
  }
}

总结

sum方法适合在比较简单的逻辑的计算中使用，ProcessFunction和RichMapFunction在实际应用环境中可以通过将状态保存到状态后端，如果出现故障通过checkpoint来恢复。