[ https://issues.apache.org/jira/browse/SPARK-11583?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14996206#comment-14996206 ]
Kent Yao commented on SPARK-11583: ---------------------------------- 1. The _Compressed_ in MapStatus means *Array[Long]* -> *Array[Byte]* -> *Long(1)* + BitSet(Array.length), which represents RealSize -> CompressedMapStatus -> HiglyCompressedMapStatus, this is the in-memory representation. Roaringbitmap is same as the BitSet now we use in HiglyCompressedMapStatus, but take 20% memory usage more than BitSet. They both don't be compressed in-memory. According to the annotations of the former Roaring-HiglyCompressedMapStatus, it can be compressed during serialization not in-memory. 2. I will try optimize my code. Maybe only need more constructors for HiglyCompressedMapStatus but a new class. 3. The mapstatuses will be fetched by executors during next stage, if the size > 0,it will fetch shuffle results produced by the former stage, if < 0, not. If we can use much small mapstatus to realize this function, it can advoid driver's oom and reduce nio consumption. Here we separate two cases(sparse/dense), use little amount of Ints which means reduceIds where the blocks belongs, as long as reduceIds[nonEmptyBlocks] or reduceId[emptyBlocks] takes less bits than BitSet[TotalBlocks], we can use this optimization. The thresholds here are 1/32 and 31/32, cases between 1/32~31/32 BitSet is smaller. Here 32 represents sizeof(Int). > Make MapStatus use less memory uage > ----------------------------------- > > Key: SPARK-11583 > URL: https://issues.apache.org/jira/browse/SPARK-11583 > Project: Spark > Issue Type: Improvement > Reporter: Kent Yao > > In the resolved issue https://issues.apache.org/jira/browse/SPARK-11271, as I > said, using BitSet can save ≈20% memory usage compared to RoaringBitMap. > For a spark job contains quite a lot of tasks, 20% seems a drop in the ocean. > Essentially, BitSet uses long[]. For example a BitSet[200k] = long[3125]. > So if we use a HashSet[Int] to store reduceId (when non-empty blocks are > dense,use reduceId of empty blocks; when sparse, use non-empty ones). > For dense cases: if HashSet[Int](numNonEmptyBlocks).size < > BitSet[totalBlockNum], I use MapStatusTrackingNoEmptyBlocks > For sparse cases: if HashSet[Int](numEmptyBlocks).size < > BitSet[totalBlockNum], I use MapStatusTrackingEmptyBlocks > sparse case, 299/300 are empty > sc.makeRDD(1 to 30000, 3000).groupBy(x=>x).top(5) > dense case, no block is empty > sc.makeRDD(1 to 9000000, 3000).groupBy(x=>x).top(5) -- This message was sent by Atlassian JIRA (v6.3.4#6332) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org