flink AggregateFunction 累加器慢

Question

我正在使用 flink 进行数据聚合 类似的代码

.keyby()
.timewidow()
.aggregate(new AggCount(), new AggWindow())


class AggCountNew() extends AggregateFunction[((String, String, Int), Message), OutMessage, OutMessage] {

  private val logger = LoggerFactory.getLogger(getClass)

  override def createAccumulator(): OutMessage = OutMessage(null, null, 0, new mutable.OpenHashMap[String, Long](128), new mutable.OpenHashMap[String, Long](32768), Set(), Set())

  override def add(value: ((String, String, Int), OutMessage), accumulator: OutMessage): OutMessage = {
    val dataMap= accumulator.dataMap
    dataMap(value._2.device) = dataMap.getOrElse[Long](value._2.device, 0) + 1
    accumulator
  }

  override def getResult(accumulator: HbaseOneDayMessage): HbaseOneDayMessage = {
    return accumulator
  }

  override def merge(a: Message, b: Message): HbaseOneDayMessage = {
    a
  }
}

当dataMap有超过1000个key时，吞吐量很低

Answer 1

您可以尝试增加 flink 托管内存。第二个参数是内存大小。

config.setString("taskmanager.memory.managed.size", "8g")
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)