[HUDI-1688] Uncache Rdd once write operation is complete

[xiarixiaoyao]

Tips

• Thank you very much for contributing to Apache Hudi.
• Please review https://hudi.apache.org/contributing.html before opening a pull request.

What is the purpose of the pull request

fix the bug that hudi cannot uncache rdds correctly
now， hudi improve write performance by cache necessary rdds； however when the write operation is finnished， those cached rdds have not been uncached which waste lots of memory.

https://github.com/apache/hudi/blob/master/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java#L115

https://github.com/apache/hudi/blob/master/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java#L214

In our environment：

step1： insert 100GB data into hudi table by spark （ok）

step2： insert another 100GB data into hudi table by spark again （oom ）

Brief change log

uncached rdds， when the write operation is finished

Verify this pull request

Existing UT

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[xiarixiaoyao]

cc @garyli1019 @nsivabalan , could you help me review this pr, thanks

[codecov-io]

Codecov Report

Merging #2673 (2ecca39) into master (e93c6a5) will increase coverage by 0.01%.
The diff coverage is 85.71%.

@@             Coverage Diff              @@
##             master    #2673      +/-   ##
============================================
+ Coverage     51.99%   52.00%   +0.01%     
  Complexity     3580     3580              
============================================
  Files           466      466              
  Lines         22275    22282       +7     
  Branches       2374     2375       +1     
============================================
+ Hits          11581    11587       +6     
- Misses         9686     9687       +1     
  Partials       1008     1008              

Flag	Coverage Δ	Complexity Δ
hudicli	37.01% <ø> (ø)	0.00 <ø> (ø)
hudiclient	100.00% <ø> (ø)	0.00 <ø> (ø)
hudicommon	51.49% <ø> (ø)	0.00 <ø> (ø)
hudiflink	53.57% <ø> (ø)	0.00 <ø> (ø)
hudihadoopmr	33.44% <ø> (ø)	0.00 <ø> (ø)
hudisparkdatasource	69.91% <85.71%> (+0.07%)	0.00 <0.00> (ø)
hudisync	49.62% <ø> (ø)	0.00 <ø> (ø)
huditimelineservice	64.36% <ø> (ø)	0.00 <ø> (ø)
hudiutilities	69.48% <ø> (ø)	0.00 <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ	Complexity Δ
...n/scala/org/apache/hudi/HoodieSparkSqlWriter.scala	52.15% <85.71%> (+0.79%)	0.00 <0.00> (ø)

[nsivabalan]

@xiarixiaoyao : thanks for your contribution. Were you able to test out the fix in your env. That subsequent writes don't incur OOMs.

[xiarixiaoyao]

@nsivabalan yes， since the problem of company information security, i cannot paste screenshot of test result and dump.
before fix
env: (executor 4 core 8G)*50
step1: merge(df, 800 , "hudikey", "testOOM", DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 616s
step2: merge(df, 800 , "hudikey", "testOOM1", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 710s
step3: merge(df, 800 , "hudikey", "testOOM2", DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 676s
step4: merge(df, 800 , "hudikey", "testOOM3", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 1077s
step5: merge(df, 800 , "hudikey", "testOOM4", DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 1154s
step6: merge(df, 800 , "hudikey", "testOOM5", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 2055s (some executor oom)

Analysis dump: we find More than 90 percent of memory is consumed by cached rdd

after fix
step1: merge(df, 800 , "hudikey", "testOOM", DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 632s
step2: merge(df, 800 , "hudikey", "testOOM1", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 710s
step3: merge(df, 800 , "hudikey", "testOOM2", DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 698s
step4: merge(df, 800 , "hudikey", "testOOM3", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 723s
step5: merge(df, 800 , "hudikey", "testOOM4", DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 616s
step6: merge(df, 800 , "hudikey", "testOOM5", DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, op = "insert") time cost: 703s

One last point， when we cached some rdds， we should uncache those rdds timely once those rdds are not used。 spark can uncached rdds automaticly but this process is uncertain。

[rubenssoto]

Great fix, In my pipeline I processed many tables in the same spark context using threads, so, probably until the end of my pipelines I have many rdd cached.

[vinothchandar]

Makes sense to explicitly unpersist(). Do you think the issue is a leak of some sort, where we are hanging onto the writeStatus RDDs? or simply the spark automatic cleaning not keeping up?
https://spark.apache.org/docs/latest/rdd-programming-guide.html#removing-data

I ask coz, it seems like you don't use blocking=true, so its all async unpersisting anyway .

[vinothchandar]

I ll land this later today. Change itself LGTM.

[xiarixiaoyao]

@rubenssoto spark will uncache rdd automaticly in a LRU fashion, so if your program has enough memory , there will be no problem. In my program hudi cached too large data in memory, program occur oom before spark automatic clean cached rdd.
@vinothchandar no leak of sort occur, simply the spark automatic cleaning not keeping up . i test on my env even if we donnot use blocking=true , spark can uncache rdd in a very short time