[
https://issues.apache.org/jira/browse/FLINK-9506?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16510704#comment-16510704
]
swy commented on FLINK-9506:
----------------------------
Hi [~sihuazhou] the picture is captured without any line being commented-out.
Yes the comment-out means no code in onTimer(). If we further comment out
recordStore.add() then everything works well, no more fluctuation, just the
input will be stopped if we output something in the processElement() instead(is
this something expected?). 16 parallelism is used. And rocksdb backend is
loaded successfully as mentioned in the log. Any optimization could be done on
Rocksdb? Or any suggestion on hashCode? Is there a way to generate a unique
hash? thank you.
> Flink ReducingState.add causing more than 100% performance drop
> ---------------------------------------------------------------
>
> Key: FLINK-9506
> URL: https://issues.apache.org/jira/browse/FLINK-9506
> Project: Flink
> Issue Type: Improvement
> Affects Versions: 1.4.2
> Reporter: swy
> Priority: Major
> Attachments: KeyNoHash_VS_KeyHash.png, flink.png,
> input_stop_when_timer_run.png, keyby.png
>
>
> Hi, we found out application performance drop more than 100% when
> ReducingState.add is used in the source code. In the test checkpoint is
> disable. And filesystem(hdfs) as statebackend.
> It could be easyly reproduce with a simple app, without checkpoint, just
> simply keep storing record, also with simple reduction function(in fact with
> empty function would see the same result). Any idea would be appreciated.
> What an unbelievable obvious issue.
> Basically the app just keep storing record into the state, and we measure how
> many record per second in "JsonTranslator", which is shown in the graph. The
> difference between is just 1 line, comment/un-comment "recStore.add(r)".
> {code}
> DataStream<String> stream = env.addSource(new GeneratorSource(loop);
> DataStream<JSONObject> convert = stream.map(new JsonTranslator())
> .keyBy()
> .process(new ProcessAggregation())
> .map(new PassthruFunction());
> public class ProcessAggregation extends ProcessFunction {
> private ReducingState<Record> recStore;
> public void processElement(Recordr, Context ctx, Collector<Record> out) {
> recStore.add(r); //this line make the difference
> }
> {code}
> Record is POJO class contain 50 String private member.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
