Как преобразовать fold () в AggregateFunction в Flink для WindowFunction? - PullRequest
Новая
       27

Как преобразовать fold () в AggregateFunction в Flink для WindowFunction?

0 голосов
/ 12 мая 2019

Я пытался преобразовать старую версию теста потоковой передачи Yahoo для Flink в новую версию, удалив устаревшие классы.

Сейчас я застрял в преобразовании устаревшей кратности () в агрегат (). Я не мог сопоставить существующие параметры сгиба с теми в совокупности. 

//old version using fold
 val windowedCounts = windowedEvents.fold(new WindowedCount(null, "", 0, new java.sql.Timestamp(0L)),
          (acc: WindowedCount, r: (String, String, Timestamp)) => {
            val lastUpdate = if (acc.lastUpdate.getTime < r._3.getTime) r._3 else acc.lastUpdate
            acc.count += 1
            acc.lastUpdate = lastUpdate
            acc
          },
          (key: Tuple, window: TimeWindow, input: Iterable[WindowedCount], out: Collector[WindowedCount]) => {
            val windowedCount = input.iterator.next()
            println(windowedCount.lastUpdate)
            out.collect(new WindowedCount(new java.sql.Timestamp(window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
            //out.collect(new WindowedCount(new java.sql.Timestamp(window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
          }
        )

val windowedCounts = windowedEvents.aggregate (new CountAggregate)

Я хочу создать класс CountAggregate, расширив класс AggregateFunction (что-то вроде): 

class CountAggregate extends AggregateFunction[(String, String, Timestamp), WindowedCount, Collector[WindowedCount]] {
    override def createAccumulator() = WindowedCount(null, "", 0, new java.sql.Timestamp(0L))

    override def accumulate(acc: WindowedCount, r: (String, String, Timestamp)): WindowedCount = {
      val lastUpdate = if (acc.lastUpdate.getTime < r._3.getTime) r._3 else acc.lastUpdate
      acc.count += 1
      acc.lastUpdate = lastUpdate
      acc
          }

    override def getValue (acc: WindowedCount)  = { (key: Tuple, window: TimeWindow, input: Iterable[WindowedCount], out: Collector[WindowedCount]) =>
      val windowedCount = input.iterator.next()
      println(windowedCount.lastUpdate)
      out.collect(new WindowedCount(new java.sql.Timestamp(window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
    }

Буду признателен за любую помощь в переписывании класса CountAggregate.

1 Ответ

1 голос
/ 14 мая 2019

Вам нужно указать AggregateFunction, а также ProcessWindowFunction, чтобы сделать последний getValue шаг: 

val windowedCounts = windowedEvents.aggregate(
      new CountAggregate(),
      new WindowAggregateFunction())

class CountAggregate extends AggregateFunction[(String, String, Timestamp), WindowedCount, WindowedCount] {
  override def createAccumulator() = WindowedCount(null, "", 0, new java.sql.Timestamp(0L))

  override def add(value: (String, String, Timestamp), acc: WindowedCount): WindowedCount = {
    val lastUpdate = if (acc.lastUpdate.getTime < value._3.getTime) value._3 else acc.lastUpdate
    WindowedCount(null, "", acc.count + 1, lastUpdate)
  }

  override def getResult(accumulator: WindowedCount): WindowedCount = {
    accumulator
  }

  override def merge(a: WindowedCount, b: WindowedCount): WindowedCount = {
    WindowedCount(null, "", a.count + b.count, if (a.lastUpdate.getTime < b.lastUpdate.getTime) b.lastUpdate else a.lastUpdate)
  }
}

class WindowAggregateFunction extends ProcessWindowFunction[WindowedCount, WindowedCount, Tuple, TimeWindow]() {
  override def process(key: Tuple, context: Context, elements: Iterable[WindowedCount], out: Collector[WindowedCount]): Unit = {
    val windowedCount = elements.iterator.next()
    out.collect(WindowedCount(new java.sql.Timestamp(context.window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
  }
}
...