本文大纲
1 RichFunction
2 RichMapFunction
3 state
4 keyed state
4.1 keyedState简单示例
4.2 利用keyedState实现流数据全局范围分组求topN

在前边flink DataSource和DataSink博文中我们或多或少都接触到了RichFunction“富函数”，只了解“富函数”中的open(),close()这两个方法，没有进一步的了解。那么在本博文我们就一起来对“富函数”进行一个了解。

1 RichFunction

之前我们第一次看到“富函数”是在自定义DataSource时，通过继承RichSourceFunction这个抽象函数类实现从mysql中读取记录到flink中。flink暴露给我们实现自定义DataSource明明需要我们传递的是一个“SourceFunction类型”，那么为什么我们继承“RichSourceFunction类型”可以实现自定义DataSoucre呢？原来是因为“RichSourceFunction类”它也是“SourceFunction类”的子类，将继承于它的子类实例作为参数传入到env.addSource()中当然可以实现自定义DataSource的目的。

这样干说似乎有点乱，也不能够让大家清楚明白其中的原因。那么大家看一下下边的图，该图将RichSourceFunction的继承关系都展现出来。

01_extends.png

可能有些老哥想知道如何得到这样的图，难道要自己画吗。当然不是，强大的IDEA早就提供了展现一个类继承关系层次结构图的功能。只需要我们将鼠标\放在我们想要了解的类文件上右击--Diagrams--Show Diagram就能得到这样的图了。而且我们还能够看到层次结构关系图上的到每一个类或接口上的方法、字段等信息。

02_show_disgram.png

通过下图，我们又能够得到一个信息，RichSourceFunction中的open(),close()等方法最终都是从“RichFunction”函数接口中继承下来的。

03_extends_and_method.png

其实我们在上一篇博文中实现自定DataSink时通过继承RichSourceFunction类得到的open(),close()方法也是从RichFunction函数接口中继承下来的。不止“RichSourceFunction类”、“RichSinkFunction类”这两个函数类是从RichFunction继承到open(),close()方法，可以说我们后边所有用到的flink提供的“富函数”中的open(),close()当然还有超级重要的“runtimeContext运行时上下文”都是RichFunction带来给我们的。

flink中RichFuntion的子类特别多，source,transformation,sink中各算子所需的编程接口参数都有继承或者间接继承RichFuntion。

04_rich_functions.png

2 RichMapFunction

现在我们就通过一个简单例子来使用一下“富函数”，这儿我们就通过RichMapFunction例子来了解下吧。下边代码展示的是flink流处理程序实现从kafka中消费数据然后通过简单的map算子转化操作并将结果打印在控制台。

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 从kafka中获取记录
    val originalStream: DataStream[String] =
      env.addSource(new FlinkKafkaConsumer010[String]("test", new SimpleStringSchema(), initProps()))

    originalStream
      .map(new MyRichMapFunction()) // 应用于每条记录
      .print()

    env.execute("rich_map")
  }

class MyRichMapFunction extends RichMapFunction[String, String] {
  var startTime: Long = _

  override def open(parameters: Configuration): Unit = {
    startTime = System.currentTimeMillis()
  }

  override def map(in: String): String = {
    // 每条记录的处理时间
    val str: String = in + "处理时间:" + System.currentTimeMillis()
    s"开始时间:$startTime, 当前数据$str"
  }

  override def close(): Unit = {}
}

05_rich_map_demo.png

上图告诉我们open()确实在RichMapFunction子类创建对象的时候只执行了一次，而且在map()方法前被调用。

3 state

flink中的state(状态)是个什么东西呢，为什么说flink能够很好的支持有状态的计算。

1.state指的是由一个任务维护并且用来计算某个结果的所有数据都属于这个状态
2.可以简单的认为state就是一个本地变量，可以被任务的业务逻辑访问(流中的数据当然也是一个个变量)
3.flink会进行状态管理，包括状态一致性、故障处理以及高效存储
(状态一致性和故障处理后边博文写)

在flink中，状态始终与特定算子相关联，毕竟一个任务的执行都是在一个个算子上完成的。当然在使用一个算子的状态前，我们需要先注册该算子的状态，否则无法使用该算子的状态。
从类型上来区分，有两种状态。官方文档中文版

1.键控状态(keyed state)：Keyed State 通常和 key 相关，仅可使用在 KeyedStream 的方法和算子中。你可以把 Keyed State 看作分区或者共享的 Operator State, 而且每个 key 仅出现在一个分区内。 逻辑上每个 keyed-state 和唯一元组 <算子并发实例, key> 绑定，由于每个 key 仅”属于” 算子的一个并发，因此简化为 <算子, key>。Keyed State 会按照 Key Group 进行管理。Key Group 是 Flink 分发 Keyed State 的最小单元； Key Group 的数目等于作业的最大并发数。在执行过程中，每个 keyed operator 会对应到一个或多个 Key Group。

2.算子状态(operator state)：对于 Operator State (或者 non-keyed state) 来说，每个 operator state 和一个并发实例进行绑定。 Kafka Connector是 Flink 中使用 operator state 的一个很好的示例。 每个 Kafka 消费者的并发在 Operator State 中维护一个 topic partition 到 offset 的映射关系。Operator State 在 Flink 作业的并发改变后，会重新分发状态，分发的策略和 Keyed State 不一样。

4 keyed state

官网描述：keyed state 接口提供不同类型状态的访问接口，这些状态都作用于当前输入数据的 key 下。换句话说，这些状态仅可在 KeyedStream 上使用，可以通过 stream.keyBy(...) 得到 KeyedStream。

• ValueState<T>: 保存一个可以更新和检索的值（如上所述，每个值都对应到当前的输入数据的 key，因此算子接收到的每个 key 都可能对应一个值）。 这个值可以通过 update(T) 进行更新，通过 T value() 进行检索。
• ListState<T>: 保存一个元素的列表。可以往这个列表中追加数据，并在当前的列表上进行检索。可以通过 add(T) 或者 addAll(List<T>) 进行添加元素，通过 Iterable<T> get() 获得整个列表。还可以通过 update(List<T>) 覆盖当前的列表。
• ReducingState<T>: 保存一个单值，表示添加到状态的所有值的聚合。接口与 ListState 类似，但使用 add(T) 增加元素，会使用提供的 ReduceFunction 进行聚合。
• AggregatingState<IN, OUT>: 保留一个单值，表示添加到状态的所有值的聚合。和 ReducingState 相反的是, 聚合类型可能与 添加到状态的元素的类型不同。 接口与 ListState 类似，但使用 add(IN) 添加的元素会用指定的 AggregateFunction 进行聚合。
• FoldingState<T, ACC>: 保留一个单值，表示添加到状态的所有值的聚合。 与 ReducingState 相反，聚合类型可能与添加到状态的元素类型不同。 接口与 ListState 类似，但使用add（T）添加的元素会用指定的 FoldFunction 折叠成聚合值。
• MapState<UK, UV>: 维护了一个映射列表。 你可以添加键值对到状态中，也可以获得反映当前所有映射的迭代器。使用 put(UK，UV) 或者 putAll(Map<UK，UV>) 添加映射。 使用 get(UK) 检索特定 key。 使用 entries()，keys() 和 values() 分别检索映射、键和值的可迭代视图。

所有类型的状态还有一个clear() 方法，清除当前 key 下的状态数据，也就是当前输入元素的 key。

注意 FoldingState 和 FoldingStateDescriptor 从 Flink 1.4 开始就已经被启用，将会在未来被删除。 作为替代请使用 AggregatingState 和 AggregatingStateDescriptor。

4.1 keyedState简单示例

下边就通过一个简单的例子来使用ValueState。该例子的目的是统计数据流中当前奇偶数总和，并打印在控制台上。

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val originalStream: DataStream[String] =
      env.addSource(new FlinkKafkaConsumer010[String]("test", new SimpleStringSchema(), initProps()))

    originalStream
      .flatMap(_.split(","))
      .filter(!_.isEmpty)
      .map { number =>
        val num: Int = number.toInt
        if (num % 2 == 0) ("even", num) else ("odd", num)
      }
      .keyBy(tp => tp._1)
      .process(new MyAverageComputer())
      .print()

    env.execute("keyby_state")
  }

class MyAverageComputer extends KeyedProcessFunction[String, (String, Int), String] {
  // keyed state: 每个实例和一个key分区对应。此处注册ValueState类型状态
  lazy val sum: ValueState[Int] =
    getRuntimeContext.getState(new ValueStateDescriptor[Int]("sum", classOf[Int]))
  override def processElement(i: (String, Int),
                              ctx: KeyedProcessFunction[String, (String, Int), String]#Context,
                              collector: Collector[String]): Unit = {
    var value: Int = sum.value() // 获取上一次的聚合结果
    value = value + i._2
    sum.update(value) //将本次的聚合结果更新回 ValueState实例
    // 收集记录到流
    collector.collect(s"${i._1}_sum: $value")
  }
}

该类中，我们通过getRuntimeContext.getState()方法和ValueStateDescriptor描述信息注册了一个ValueState[Int]状态。利用该状态在每次对于当前记录进行操作前都会把上次的累加结果取出来和本次记录进行累加，再把累加结果更新回去。相当于我们利用了一个变量，保存每次的累加结果。(等等，这样理解的话，那我们为什么不直接用个变量来保存不就得了嘛，干嘛还非要通过runtimeContext和state做个中间商赚差价呢？这么做是为了容灾和故障恢复的时候能保证状态不会被更改，runtime会对这些状态进行编码并写入checkpoint)

06_keyed_state.png

4.2 利用keyedState实现流数据全局范围分组求top N

现在我们遇到了这么一个场景：需要统计出当前商城历史到现在各个城市销量最高的前N种商品。这是一个典型的分组求top n的问题，只不过这儿数据范围是流数据全局。

我们的处理过程思路主要分为两大步骤

1.按照 城市city+商品itemId 先作为一个key。
  使用keyBy算子进行分组，并且对每一个分组内商品的销量进行累加统计
2.将步骤1中的结果再根据city进行分组，此时每一个分组内的数据都是同一个city的不同商品销量。
  然后我们再同一个city分组内对商品按照销量进行排序，只取前N个商品

步骤1的实现代码

// 订单详情
case class OrderDesc(orderId: String, itemId: String, count: Int, city: String, createTime: Long)
// 商品销售累计结果
case class OrderCount(itemId: String, city: String, sum: Long, time: Long)

    val originalStream: DataStream[String] = ...

    val baseStream: DataStream[OrderCount] = originalStream
      .filter(!_.isEmpty) // 去除空字符串
      .map { line =>
        val strs: Array[String] = line.split(",")
        OrderDesc(strs(0), strs(1), strs(2).toInt, strs(3), strs(4).toLong) // 解析字符串生成OrderDesc对象
      }
      .keyBy(entry => entry.city + "|" + entry.itemId) // 根据 城市和商品名 进行分组
      .process(new BaseSalesSumProcess())

// 不断累加数据流中 城市和商品 的销售总数，然后输出到下游。
class BaseSalesSumProcess extends KeyedProcessFunction[String, OrderDesc, OrderCount] {
  var salesSum: ValueState[Long] = _    // 声明一个键控状态ValueState

  override def open(parameters: Configuration): Unit = {
    // 根据描述信息注册ValueState，描述信息中设置ValueState的value默认值为0
    salesSum = getRuntimeContext.getState(new ValueStateDescriptor[Long]("salesSum", classOf[Long], 0L))
  }

  override def processElement(in: OrderDesc, ctx: KeyedProcessFunction[String, OrderDesc, OrderCount]#Context,
                              collector: Collector[OrderCount]): Unit = {
    // 获取状态中的数据
    val oldSum: Long = salesSum.value()
    val newSum = oldSum + in.count
    // 更新状态中的数据
    salesSum.update(newSum)

    // 收集记录到流中。将最新的订单时间作为统计结果的时间
    collector.collect(OrderCount(in.itemId, in.city, newSum, in.createTime))
  }
}

原始数据经过上边处理，我们已经能够得到 city+itemId 维度的商品销量数据。并且是不断更新输出到下游的，这个时候我们就需要继续进行 city 维度的商品销量top n的计算。想要得到top n的数据，其实就是对 city 维度内的商品按照销量大小进行倒序排序，然后只取前n个。首先我们想到的是可以使用堆排序来达到这个目的，先构建一个小根堆，然后再将小根堆进行一个排序操作。这儿我们就不自己写代码来实现这个功能了，直接使用TreeSet来帮助实现类似的功能。

步骤2实现代码

    // baseStrem就是经过步骤1处理后的数据流
    baseStream.keyBy(entry => entry.city)
      .process(new TopnProcess())
      .print()

// 判断 商品 的销售总数是否在top N中，并将当下的top N数据输出到下游。
class TopnProcess extends KeyedProcessFunction[String, OrderCount, String] {
  var topN: ValueState[java.util.TreeSet[OrderCount]] = _
  var map: MapState[String, OrderCount] = _

  override def open(parameters: Configuration): Unit = {
    // 注册一个ValueState，实现 优先级队列功能(存放top n的数据)
    topN = getRuntimeContext.getState(
      new ValueStateDescriptor[util.TreeSet[OrderCount]]("topN", classOf[util.TreeSet[OrderCount]]))
    // 将itemId和OrderCount关系记录下来，用来判断在 topN 中是否存有该OrderCount
    map = getRuntimeContext.getMapState(new MapStateDescriptor[String, OrderCount]("map", classOf[String], classOf[OrderCount]))
  }

  override def processElement(in: OrderCount,
                              ctx: KeyedProcessFunction[String, OrderCount, String]#Context,
                              collector: Collector[String]): Unit = {
    var records: util.TreeSet[OrderCount] = topN.value()
    // top n的构建过程
    // 判断records是否为null，为null表明该数据一定符合top n条件
    if (records == null) {
      // 升序比较
      records = new util.TreeSet[OrderCount](new Comparator[OrderCount] {
        override def compare(o1: OrderCount, o2: OrderCount): Int = {
          val value: Int = (o1.sum - o2.sum).toInt
          // 避免当两个OrderCount中的sum值都一样时第二个OrderCount无法添加到TreeSet中
          if (value == 0)
            o1.itemId.hashCode - o2.itemId.hashCode
          else value

        }
      })
      records.add(in) // 添加OrderCount到top N中
      map.put(in.itemId, in) // 更新map记录

    } else {
      map.contains(in.itemId) match {
        // 存在
        case true =>
          val old: OrderCount = map.get(in.itemId)
          // 替换并更新map记录
          if (in.sum >= old.sum) {
            records.remove(old)
            records.add(in)
            map.put(in.itemId, in)// 更新map记录
          }
        case false =>
          // 不存在，判断是否已经到了top n的size
          if (records.size >= 3) {
            val min: OrderCount = records.first()
            // 新的值更大则替换原来最小的数据
            if (in.sum >= min.sum) {
              records.pollFirst()   // 移除最小的数据
              records.add(in)
              // 更新map记录
              map.remove(min.itemId)
              map.put(in.itemId, in)
            }
          } else {
            // 未到top n的size，直接新增
            records.add(in)
            map.put(in.itemId, in)// 更新map记录
          }
      }
    }

    // 将结果存放到ValueState中
    topN.update(records)
    // 将当前的top N数据收集到流中
    collector.collect("------------------start---------------------")
    map.entries().forEach { entry =>
      collector.collect(s"key:${entry.getKey}, value:${entry.getValue.toString}")
    }
    records.forEach { record =>
      collector.collect(s"city:${record.city}, item:${record.itemId}, sum:${record.sum}")
    }
    collector.collect("-------------------end----------------------")
  }
}

整体逻辑就是：

判断一个itemId的对象是否已经存在TreeSet中？
   已经存在，那么itemId的新销售数量是否大于原TreeSet中的，大于的话新的更新就替换旧的。
   如果不存在，就要需要先判断TreeSet的size是否已经到了n？
      没有到n则直接新增。
      到了n就和原TreeSet中的最小的itemId的销售数量进行比较大小，新的更大就去替换。

此处我们不仅注册了var topN: ValueState[java.util.TreeSet[OrderCount]]还注册了var map: MapState[String, OrderCount] = _。ValueState[java.util.TreeSet[T]]帮助我们实现了top n的计算，MapState[K,V]则是帮助我们判断TreeSet原来是否已经存在一个itemId的OrderCount对象示例，如果存在是否要替换该对象。

07_global_top_n.png

当然，关于RichFunction和state的知识肯定不止这么一点，本博文只是讲了一点点，并且通过两个例子来说明了一下KeyBy State。具体的更详细的State知识大家去参考官网的。官方文档中文版，关于state的介绍