我有一个Flink星系团。我启用了压缩过滤器并使用状态ttl。但是rocksdb压缩过滤器不能从内存中释放状态。
我的flink管道里有大约300张唱片
我的状态ttl配置:
@Override
public void open(Configuration parameters) throws Exception {
ListStateDescriptor<ObjectNode> descriptor = new ListStateDescriptor<ObjectNode>(
"my-state",
TypeInformation.of(new TypeHint<ObjectNode>() {})
);
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(600))
.cleanupInRocksdbCompactFilter(2)
.build();
descriptor.enableTimeToLive(ttlConfig);
myState = getRuntimeContext().getListState(descriptor);
}
flink-conf.yaml公司:
state.backend: rocksdb
state.backend.rocksdb.ttl.compaction.filter.enabled: true
state.backend.rocksdb.block.blocksize: 16kb
state.backend.rocksdb.compaction.level.use-dynamic-size: true
state.backend.rocksdb.thread.num: 4
state.checkpoints.dir: file:///opt/flink/checkpoint
state.backend.rocksdb.timer-service.factory: rocksdb
state.backend.rocksdb.checkpoint.transfer.thread.num: 2
state.backend.local-recovery: true
state.backend.rocksdb.localdir: /opt/flink/rocksdb
jobmanager.execution.failover-strategy: region
rest.port: 8081
state.backend.rocksdb.memory.managed: true
# state.backend.rocksdb.memory.fixed-per-slot: 20mb
state.backend.rocksdb.memory.write-buffer-ratio: 0.9
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1
taskmanager.memory.managed.fraction: 0.6
taskmanager.memory.network.fraction: 0.1
taskmanager.memory.network.min: 500mb
taskmanager.memory.network.max: 700mb
taskmanager.memory.process.size: 5500mb
taskmanager.memory.task.off-heap.size: 800mb
metrics.reporter.influxdb.class: org.apache.flink.metrics.influxdb.InfluxdbReporter
metrics.reporter.influxdb.host: ####
metrics.reporter.influxdb.port: 8086
metrics.reporter.influxdb.db: ####
metrics.reporter.influxdb.username: ####
metrics.reporter.influxdb.password: ####
metrics.reporter.influxdb.consistency: ANY
metrics.reporter.influxdb.connectTimeout: 60000
metrics.reporter.influxdb.writeTimeout: 60000
state.backend.rocksdb.metrics.estimate-num-keys: true
state.backend.rocksdb.metrics.num-running-compactions: true
state.backend.rocksdb.metrics.background-errors: true
state.backend.rocksdb.metrics.block-cache-capacity: true
state.backend.rocksdb.metrics.block-cache-pinned-usage: true
state.backend.rocksdb.metrics.block-cache-usage: true
state.backend.rocksdb.metrics.compaction-pending: true
XDB和grafana监测:
1条答案
按热度按时间jvidinwx1#
正如这个ttl cleanup的名字所暗示的(cleanupinrocksdbcompactfilter),它依赖于定制的rocksdb压缩过滤器,它只在压缩过程中运行。更多细节见文档。
屏幕截图中的指标显示一直没有运行压缩。我认为数据的大小不足以在此时开始任何压缩。
压缩过滤器不会从内存中释放状态。
我假设主ram内存的意思是“from memory”。如果是这样,压实根本就没有在那里进行。rocksdb在主内存中保存的数据大小总是有限的。它基本上是一个缓存,过期的未触及状态最终应该被从缓存中逐出。其余的部分会周期性地溢出到磁盘中,并随着时间的推移而被压缩。此时ttl清理应该从系统中删除过期状态。