State的作用
state是Flink程序某个时刻某个task/operator的状态,state数据是程序运行中某一时刻数据结果。首先要将state和checkpoint概念区分开,可以理解为checkpoint是要把state数据持久化存储起来,checkpoint默认情况下会存储在JoManager的内存中。checkpoint表示一个Flink job在一个特定时刻的一份全局状态快照,方便在任务失败的情况下数据的恢复。
State 状态值存储(checkpoint会存储在hdfs上)
env.setStateBackend(new FsStateBackend("hdfs:///user/flink/app_statistics/checkpoint"))
checkpoint存储state数据,重启时恢复state数据
//设置checkpoint, job失败重启可以恢复数据, 默认是CheckpointingMode.EXACTLY_ONCE
//flink-conf.yaml配置文件中配置了默认的重启策略: fixed-delay(4, 10s)
env.enableCheckpointing(60000)
//不希望因为checkpoint的失败而导致task失败
env.getCheckpointConfig.setFailOnCheckpointingErrors(false)
//设置checkpoint的存储管理
env.setStateBackend(new FsStateBackend("hdfs:///user/flink/app_statistics/checkpoint"))
State的应用
-
State->KeyedState(最常用的)
KeyedState是基于KeyedStream上的状态,这个状态是跟特定的key绑定的,对KeyedStream流上的每个key都有对应的state。Keyed State 仅仅可以被使用在基于KeyStream上的Rich functions。
案例一:Flink:Keyed State,实现蒙特卡洛模拟求Pi
重写map方法
// 定义一个MonteCarlo类
case class MonteCarloPoint(x: Double, y: Double) {
def pi = if (x * x + y * y <= 1) 1 else 0
}
object MonteCarko extends App {
// 自定义一个Source,实现随机坐标点的生成
class MonteCarloSource extends RichSourceFunction[MonteCarloPoint] {
val env = StreamExecutionEnvironment.getExecutionEnvironment
// state 需要在RichFunction中实现
val myMapFun = new RichMapFunction[(Long, MonteCarloPoint), (Long, Double)] {
// 定义原始状态
var countAndPi: ValueState[(Long, Long)] = _
override def map(value: (Long, MonteCarloPoint)): (Long, Double) = {
// 通过 ValueState.value获取状态值
val tmpCurrentSum = countAndPi.value
val currentSum = if (tmpCurrentSum != null) {
tmpCurrentSum
} else {
(0L, 0L)
}
val allcount = currentSum._1 + 1
val picount = currentSum._2 + value._2.pi
// 计算新的状态值
val newState: (Long, Long) = (allcount, picount)
// 更新状态值
countAndPi.update(newState)
//输出总样本量和模拟极速那的Pi值
(allcount, 4.0 * picount / allcount)
}
override def open(parameters: Configuration): Unit = {
countAndPi = getRuntimeContext.getState(
new ValueStateDescriptor[(Long, Long)]("MonteCarloPi", createTypeInformation[(Long, Long)])
)
}
}
// 添加数据源
val dataStream: DataStream[MonteCarloPoint] = env.addSource(new MonteCarloSource)
// 转换成KeyedStream
val keyedStream= dataStream.map((1L, _)).keyBy(0)
// 调用定义好的RichFunction并打印结果
keyedStream.map(myMapFun).print()
env.execute("Monte Carko Test")
}
案例二:官网案例(重写flatmap方法)
import java.lang
import org.apache.flink.api.common.functions.{RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector
import org.apache.flink.configuration.Configuration
import scala.collection.JavaConverters._
class CountWindowAverage extends RichFlatMapFunction[(Long, Long), (Long, Long)] {
private var sum: ValueState[(Long, Long)] = _
override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = {
// access the state value
val tmpCurrentSum = sum.value
// If it hasn't been used before, it will be null
val currentSum = if (tmpCurrentSum != null) {
tmpCurrentSum
} else {
(0L, 0L)
}
// update the count
val newSum = (currentSum._1 + 1, currentSum._2 + input._2)
// update the state
sum.update(newSum)
// if the count reaches 2, emit the average and clear the state
if (newSum._1 >= 2) {
out.collect((input._1, newSum._2 / newSum._1))
sum.clear()
}
}
override def open(parameters: Configuration): Unit = {
sum = getRuntimeContext.getState(
new ValueStateDescriptor[(Long, Long)]("average", createTypeInformation[(Long, Long)])
)
}
}
object ExampleCountWindowAverage extends App {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.fromCollection(List(
(1L, 3L),
(1L, 5L),
(1L, 7L),
(1L, 4L),
(1L, 2L)
)).keyBy(_._1)
.flatMap(new CountWindowAverage()).print()
// the printed output will be (1,4) and (1,5)
env.execute("ExampleManagedState")
}
案例三:计算最热门top3商品
ProcessFunction是Flink提供的一个low-level API,用于实现更高级的功能。它主要提供了定时器timer的功能(支持EventTime或ProcessingTime)。本案例中我们将利用timer来判断何时收齐了某个window下所有商品的点击量数据。由于Watermark的进度是全局的,在processElement方法中,每当收到一条数据ItemViewCount,我们就注册一个windowEnd+1的定时器(Flink框架会自动忽略同一时间的重复注册)。windowEnd+1的定时器被触发时,意味着收到了windowEnd+1的Watermark,即收齐了该windowEnd下的所有商品窗口统计值。我们在onTimer()中处理将收集的所有商品及点击量进行排序,选出TopN,并将排名信息格式化成字符串后进行输出。
这里我们还使用了ListState<ItemViewCount>来存储收到的每条ItemViewCount消息,保证在发生故障时,状态数据的不丢失和一致性。ListState是Flink提供的类似Java List接口的State API,它集成了框架的checkpoint机制,自动做到了exactly-once的语义保证。
import com.sun.jmx.snmp.Timestamp
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
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.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import scala.collection.mutable.ListBuffer
case class UserBehavior(userId: Long, itemId: Long, categoryId: Int,
behavior: String, timestamp: Long)
case class ItemViewCount(itemId: Long, windowEnd: Long, count: Long)
object UserBehaviorAnalysis {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.setParallelism(1)
val value: DataStream[UserBehavior] = env.readTextFile("D:\\projects\\flinkStudy\\src\\userBehavior.csv").
map(line => {
val linearray = line.split(",")
UserBehavior(linearray(0).toLong, linearray(1).toLong, linearray(2).toInt, linearray(3), linearray(4).toLong)
})
val watermarkDataStream: DataStream[UserBehavior] = value.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[UserBehavior]
(Time.milliseconds(1000)) {
override def extractTimestamp(element: UserBehavior): Long = {
return element.timestamp
}
})
val itemIdWindowStream: DataStream[ItemViewCount] = watermarkDataStream.filter(_.behavior == "pv").
keyBy("itemId").
timeWindow(Time.minutes(60),Time.minutes(5))
//按照每个窗口进行聚合
.aggregate(new CountAgg(), new WindowResultFunction())
itemIdWindowStream.keyBy("windowEnd").process(new TopNHotItems(3)).print()
env.execute("Hot Items Job")
}
}
class CountAgg extends AggregateFunction[UserBehavior, Long, Long] {
override def createAccumulator(): Long = 0L
override def add(userBehavior: UserBehavior, acc: Long): Long = acc + 1
override def getResult(acc: Long): Long = acc
override def merge(acc: Long, acc1: Long): Long = acc1+acc
}
// 用于输出窗口的结果
class WindowResultFunction extends WindowFunction[Long, ItemViewCount, Tuple, TimeWindow] {
override def apply(key: Tuple, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
var itemId=key.asInstanceOf[Tuple1[Long]]._1
var count=input.iterator.next()
out.collect(ItemViewCount(itemId, window.getEnd, count))
}
}
class TopNHotItems(topSize: Int) extends KeyedProcessFunction[Tuple, ItemViewCount, String] {
private var itemState : ListState[ItemViewCount] = _
override def open(parameters: Configuration): Unit = {
super.open(parameters)
// 命名状态变量的名字和状态变量的类型
val itemsStateDesc = new ListStateDescriptor[ItemViewCount]("itemState-state", classOf[ItemViewCount])
// 定义状态变量
itemState = getRuntimeContext.getListState(itemsStateDesc)
}
override def processElement(input: ItemViewCount, context: KeyedProcessFunction[Tuple, ItemViewCount, String]#Context, collector: Collector[String]): Unit = {
// 每条数据都保存到状态中
itemState.add(input)
// 注册 windowEnd+1 的 EventTime Timer, 当触发时,说明收齐了属于windowEnd窗口的所有商品数据
// 也就是当程序看到windowend + 1的水位线watermark时,触发onTimer回调函数
context.timerService.registerEventTimeTimer(input.windowEnd + 1)
}
override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
// 获取收到的所有商品点击量
val allItems: ListBuffer[ItemViewCount] = ListBuffer()
import scala.collection.JavaConversions._
for (item <- itemState.get) {
allItems += item
}
// 提前清除状态中的数据,释放空间
itemState.clear()
// 按照点击量从大到小排序
val sortedItems = allItems.sortBy(_.count)(Ordering.Long.reverse).take(topSize)
// 将排名信息格式化成 String, 便于打印
val result: StringBuilder = new StringBuilder
result.append("====================================\n")
result.append("时间: ").append(new Timestamp(timestamp - 1)).append("\n")
for(i <- sortedItems.indices){
val currentItem: ItemViewCount = sortedItems(i)
// e.g. No1: 商品ID=12224 浏览量=2413
result.append("No").append(i+1).append(":")
.append(" 商品ID=").append(currentItem.itemId)
.append(" 浏览量=").append(currentItem.count).append("\n")
}
result.append("====================================\n\n")
// 控制输出频率,模拟实时滚动结果
Thread.sleep(1000)
out.collect(result.toString)
}
}