flink 1.11.2 cep rocksdb 性能调优

Wed, 04 Nov 2020 19:06:26 -0800 

hi

我这边用flink1.11.2 cep做一些模式匹配，发现一旦开启rocksdb做为状态后端，就会出现反压。cpu使用率是之前的10倍。

private void bufferEvent(IN event, long currentTime) throws Exception {
    long currentTs = System.currentTimeMillis();
    List<IN> elementsForTimestamp =  elementQueueState.get(currentTime);
    if (elementsForTimestamp == null) {
        this.bufferEventGetNullhistogram.update(System.currentTimeMillis()
- currentTs);
        elementsForTimestamp = new ArrayList<>();
    }else {
        
this.bufferEventGethistogram.update(System.currentTimeMillis()-currentTs);
    }
    elementsForTimestamp.add(event);
    long secondCurrentTs = System.currentTimeMillis();
    elementQueueState.put(currentTime, elementsForTimestamp);
    this.bufferEventPuthistogram.update(System.currentTimeMillis() -
secondCurrentTs);
    this.bufferEventhistogram.update(System.currentTimeMillis() - currentTs);
}

通过复写CepOperator,加入了一些metics发现

this.bufferEventhistogram = metrics.histogram("buffer_event_delay", new
        DescriptiveStatisticsHistogram(1000));
this.bufferEventGethistogram = metrics.histogram("buffer_event_get_delay", new
        DescriptiveStatisticsHistogram(1000));
this.bufferEventGetNullhistogram =
metrics.histogram("buffer_event_get_null_delay", new
        DescriptiveStatisticsHistogram(1000));
this.bufferEventPuthistogram = metrics.histogram("buffer_event_put_delay", new
        DescriptiveStatisticsHistogram(1000));

在get和put比较耗时，整个bufferEvent 能达到200ms
从rocksdb的metric来看没有进行太多flush和compaction。

[image: image.png]
[image: image.png]

也参考了https://www.jianshu.com/p/2e61c2c83c57这篇文章调优过，发现效果也不是很好，一样反压。
也看过类似的问题http://apache-flink.147419.n8.nabble.com/rocksDB-td1785.html
，但是我这sst文件很小。
请教大家，为啥get和put这么耗时呢？有什么好的优化方案不？谢谢。

Best Wishes.



Best Wishes.

回复