2

我正在使用 kafka -> flink -> 弹性搜索在 java 中处理 poc 项目。

在 kafka 上，将产生不可预测的事件数量，从 0 到每秒数千个事件，例如特定主题。

{"gid":"abcd-8910-2ca4227527f9", "state":"stateA", "timestamp:1465566255, "other unusefull info":"..."}

Flink 将消耗这些事件，并且应该每秒沉入弹性搜索每个状态下的事件数量，例如：

{"stateA":54, "stateB":100, ... "stateJ":34}

我有 10 个状态：[Created, ... , Deleted]平均生命周期为 15 分钟。状态可以每秒更改两次。理论上可以添加新的状态。

为了每秒接收流，我正在考虑使用 flink 的时间窗口https://flink.apache.org/news/2015/12/04/Introducing-windows.html

问题是我需要有状态的对象，其中包含关于guid->previous-state和stateX->count的信息，以便能够在新事件发生时增加/减少计数。

我找到了一份关于有状态蒸汽处理的文档草案https://cwiki.apache.org/confluence/display/FLINK/Stateful+Stream+Processing

我是 flink 和流处理的新手，我还没有深入研究 flink 有状态流处理。对于第一阶段，我正在考虑为此使用静态对象，但是当启动多个 flink 实例时，这种方法将不起作用。

我要问你：

  1. 您如何看待这种方法？
  2. flink 适合这种流处理吗？
  3. 解决这个问题的方法是什么？

我也很欣赏窗口状态流解决方案（或其他解决方案）的一些代码片段。

谢谢，

4

1 回答 1

1

像下面这样的东西怎么样？

它使用 15 分钟的窗口，之后窗口状态将被清理。它还使用一个自定义触发器，每秒评估一次窗口。就窗口操作而言，有一个 ReduceFunction 仅保留每个 guid 的最新状态，还有一个 WindowFunction 发出 (state, 1) 元组。然后我们按此状态键入并求和。我认为这应该给你你正在寻找的结果。

val env = StreamExecutionEnvironment.getExecutionEnvironment()
val stream = env.addSource(new FlinkKafkaProducer(...))

val results = stream
  .keyBy(_.guid)
  .timeWindow(Time.minutes(15))
  .trigger(ProcessingTimeTriggerWithPeriodicFirings(1000))
  .apply(
    (e1, e2) => e2,
    (k, w, i, c: Collector[(String, Long)]) => {
      if (i.head != null) c.collect((i.head.state, 1))
    }
  )
  .keyBy(0)
  .timeWindow(Time.seconds(1))
  .sum(1)
  .addSink(new ElasticsearchSink<>(...))

env.execute("Count States")

ProcessingTimeTriggerWithPeriodicFirings 定义如下：

object ProcessingTimeTriggerWithPeriodicFirings {
  def apply(intervalMs: Long) = {
    new ProcessingTimeTriggerWithPeriodicFirings(intervalMs)
  }
}

class ProcessingTimeTriggerWithPeriodicFirings(intervalMs: Long)
  extends Trigger[Event, TimeWindow] {

  private val startTimeDesc =
    new ValueStateDescriptor[Long]("start-time", classOf[Long], 0L)

  override def onElement(element: Event, timestamp: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult = {
    val startTime = ctx.getPartitionedState(startTimeDesc)
    if (startTime.value == 0) {
      startTime.update(window.getStart)
      ctx.registerProcessingTimeTimer(window.getEnd)
      ctx.registerProcessingTimeTimer(System.currentTimeMillis() + intervalMs)
    }
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(time: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult = {
    if (time == window.getEnd) {
      TriggerResult.PURGE
    }
    else {
      ctx.registerProcessingTimeTimer(time + intervalMs)
      TriggerResult.FIRE
    }
  }

  override def onEventTime(time: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult = {
    TriggerResult.CONTINUE
  }
}
于 2016-06-23T00:47:40.607 回答