Re: Read Hive table in Stream Mode use distinct cause heap OOM

Posted by Shengkai Fang on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Read-Hive-table-in-Stream-Mode-use-distinct-cause-heap-OOM-tp43204p43221.html

Hi, could you tell me which version do you use? I just want to check whether there are any problems.

Best,

Shengkai

张颖 <[hidden email]> 于2021年4月25日周日 下午5:23写道：

hi,I met an appearance like this:

this is my sql:

SELECT distinct header,label,reqsig,dense_feat,item_id_feat,user_id_feat FROM app.app_ranking_feature_table_clk_ord_hp_new_all_tree_orc where dt='2021-04-01'

When I useBlinkPlanner inBatchMode, It works well; But if I set inStreamMode,

It cause a heap OOM.

Caused by: java.lang.OutOfMemoryError: Java heap space

    at java.util.Arrays.copyOf(Arrays.java:3236)

    at org.apache.flink.core.memory.DataOutputSerializer.getCopyOfBuffer(DataOutputSerializer.java:85)

    at org.apache.flink.contrib.streaming.state.RocksDBSerializedCompositeKeyBuilder.buildCompositeKeyNamespace(RocksDBSerializedCompositeKeyBuilder.java:113)

    at org.apache.flink.contrib.streaming.state.AbstractRocksDBState.serializeCurrentKeyWithGroupAndNamespace(AbstractRocksDBState.java:163)

    at org.apache.flink.contrib.streaming.state.RocksDBValueState.value(RocksDBValueState.java:83)

    at org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:129)

    at org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:43)

    at org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)

    at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:191)

    at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:204)

    at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:174)

    at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)

    at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:399)

    at org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$221/285424866.runDefaultAction(Unknown Source)

    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191)

    at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:620)

    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:584)

    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:844)

    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:636)

    at java.lang.Thread.run(Thread.java:748)

I use the rocksdb, and I confirm it works,then I jmap the tm:

num     #instances         #bytes  class name

----------------------------------------------

   1:        214656     4420569368  [C

   2:        111199     2376771576  [B

   3:        137904        7722624  org.apache.flink.core.memory.HybridMemorySegment

   4:        214539        5148936  java.lang.String

   5:         31796        2635104  [Ljava.lang.Object;

   6:        105133        2523192  [Lorg.apache.flink.core.memory.MemorySegment;

   7:        105115        2522760  org.apache.flink.table.data.binary.BinarySection

   8:        105115        2522760  org.apache.flink.table.data.binary.BinaryStringData

   9:         32812        2099968  java.nio.DirectByteBuffer

  10:         14838        1651560  java.lang.Class

  11:         50002        1600064  java.util.concurrent.ConcurrentHashMap$Node

  12:         43014        1376448  java.util.Hashtable$Entry

  13:         32805        1312200  sun.misc.Cleaner

It looks like the data is in heap rather than in rocksdb, Is there any way to set the data to the rocksdb?