OK,我这边加个metric,先观察下

yue ma <mayuefi...@gmail.com> 于2022年3月18日周五 12:23写道:

> hi
> 我觉得这里可以注意两地方
> 首先 你可以观察一下这个时候 task 的吞吐量是多少 ,如果 qps 特别高 ,比如作业重最旧的offset 消费,我觉得这个时候 cpu 100%
> 是符合预期的。
> 其次 你可以在代码中加一些内存缓存的逻辑 类似于 mini-batch, 来减少和 state 交互的频率,也许这样能缓解一部分问题。
>
> deng xuezhao <zhm_...@outlook.com> 于2022年3月18日周五 11:19写道:
>
> > 退订
> >
> >
> >
> > 在 Peihui He <peihu...@gmail.com>,2022年3月18日 上午11:18写道:
> >
> > Hi, all
> >
> > 如题,flink 任务使用rocksdb 做为状态后端,任务逻辑大概意思是:
> > 来一条数据先判断该数据的key 是否再mapstat 中, 然后再将该key 写入mapstat中。
> >
> > 产生问题是当数据跑一段时间后,判断是否存在线程cpu总是100%,堆栈如下:
> >
> > "process (6/18)#0" Id=80 RUNNABLE (in native)
> >     at org.rocksdb.RocksDB.get(Native Method)
> >     at org.rocksdb.RocksDB.get(RocksDB.java:2084)
> >     at
> >
> >
> org.apache.flink.contrib.streaming.state.RocksDBMapState.contains(RocksDBMapState.java:173)
> >     at
> >
> >
> org.apache.flink.runtime.state.UserFacingMapState.contains(UserFacingMapState.java:72)
> >     at
> >
> >
> com.huanju.security.soc.internal.hs.bigdata.FileScanToTiDB$$anon$12.processElement(FileScanToTiDB.scala:156)
> >     at
> >
> >
> com.huanju.security.soc.internal.hs.bigdata.FileScanToTiDB$$anon$12.processElement(FileScanToTiDB.scala:145)
> >     at
> >
> >
> org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
> >     at
> >
> >
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:233)
> >     at
> > org.apache.flink.streaming.runtime.io
> >
> .AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
> >     at
> > org.apache.flink.streaming.runtime.io
> >
> .AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
> >     at
> > org.apache.flink.streaming.runtime.io
> > .StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
> >     at
> >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:496)
> >     at
> >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$624/715942770.runDefaultAction(Unknown
> > Source)
> >     at
> >
> >
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
> >     at
> >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:809)
> >     at
> >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:761)
> >     at
> >
> org.apache.flink.runtime.taskmanager.Task$$Lambda$773/520411616.run(Unknown
> > Source)
> >     at
> >
> >
> org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
> >     at
> > org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
> >     at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
> >     at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
> >     at java.lang.Thread.run(Thread.java:748)
> >
> > 但是看checkpoint数据,才100m左右
> >
> > 请问大家 rocksdb 是出现什么性能瓶颈了呢? 改怎么调优呢?
> >
>

回复