wushengran/flinktutorial

flinktutorial's Issues

笔记的问题

武老师，源码中的笔记怎么找不到了

flink自定义udaf实现lag函数的问题，老师帮忙看一下

这是我的代码，不知道为什么就是不走accumulate方法，看老师的视频，自己写的求平均值的方法可以正常运行。请老师有时间帮忙看一下，谢谢！

 class PreviousAcc {
    var queue: util.LinkedList[Int] = _
  }

  class PreviousTemp extends AggregateFunction[Int, PreviousAcc] {

    override def getValue(accumulator: PreviousAcc): Int = {
      System.out.println("进入获取结果方法！！！")
      val i1 = accumulator.queue.size()
      System.out.println(i1)
      accumulator.queue.removeLast()
    }

    override def createAccumulator(): PreviousAcc = {
      System.out.println("进入创建方法！！！")
      val value = new PreviousAcc
      value.queue = new util.LinkedList[Int]()
      value
    }

    def accumulate(acc: PreviousAcc, value: Int, length: Int): Unit = {
      System.out.println("进入计算方法！！！")
      acc.queue.addFirst(value)
      val size = acc.queue.size
      if (size < length) {
        for (i <- size until (length + 1)) {
          acc.queue.add(0);
        }
      } else if (size > length) throw new RuntimeException("队列的大小出错了！！")
    }
  }