[HUDI-1483] Support async clustering for deltastreamer and Spark streaming

[codope]

What is the purpose of the pull request

This pull request adds async clustering support for HoodieDeltaStreamer and Spark streaming writes to Hudi table.

Brief change log

• Async clustering is configurable.
• Reuses the existing clustering methods in write client and timeline.
• Hence, it has the same limitations that exist with current clustering strategy i.e. updates are rejected.

Verify this pull request

This change added tests and can be verified as follows:

• Added unit tests in TestHoodieDeltaStreamer and TestStructuredStreaming.
• Manually verified the change by running a job locally.

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.

[hudi-bot]

CI report:

• 7d4f981 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run travis re-run the last Travis build
• @hudi-bot run azure re-run the last Azure build

[codecov-commenter]

Codecov Report

Merging #3142 (7d4f981) into master (9b01d2a) will decrease coverage by 1.70%.
The diff coverage is 49.14%.

@@             Coverage Diff              @@
##             master    #3142      +/-   ##
============================================
- Coverage     47.67%   45.97%   -1.71%     
+ Complexity     5516     4727     -789     
============================================
  Files           929      832      -97     
  Lines         41303    37953    -3350     
  Branches       4144     3811     -333     
============================================
- Hits          19692    17449    -2243     
+ Misses        19863    18887     -976     
+ Partials       1748     1617     -131     

Flag	Coverage Δ
hudicli	39.97% <ø> (ø)
hudiclient	22.89% <9.52%> (-11.68%)	⬇️
hudicommon	48.56% <0.00%> (-0.03%)	⬇️
hudiflink	60.03% <ø> (ø)
hudihadoopmr	51.29% <ø> (ø)
hudisparkdatasource	67.30% <56.66%> (-0.02%)	⬇️
hudisync	54.51% <ø> (ø)
huditimelineservice	64.07% <ø> (ø)
hudiutilities	59.26% <90.19%> (+0.69%)	⬆️

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

Impacted Files	Coverage Δ
.../org/apache/hudi/async/AsyncClusteringService.java	0.00% <0.00%> (ø)
...ava/org/apache/hudi/async/AsyncCompactService.java	0.00% <0.00%> (ø)
...java/org/apache/hudi/async/HoodieAsyncService.java	0.00% <0.00%> (ø)
...g/apache/hudi/client/AbstractClusteringClient.java	0.00% <0.00%> (ø)
...java/org/apache/hudi/config/HoodieWriteConfig.java	42.80% <0.00%> (-0.08%)	⬇️
...a/org/apache/hudi/common/util/ClusteringUtils.java	88.40% <0.00%> (-1.31%)	⬇️
...in/scala/org/apache/hudi/HoodieStreamingSink.scala	28.00% <36.66%> (+4.00%)	⬆️
...n/scala/org/apache/hudi/HoodieSparkSqlWriter.scala	72.03% <72.00%> (+0.76%)	⬆️
...org/apache/hudi/config/HoodieClusteringConfig.java	71.28% <75.00%> (+0.01%)	⬆️
...apache/hudi/utilities/deltastreamer/DeltaSync.java	71.56% <75.00%> (+0.42%)	⬆️
... and 110 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 9b01d2a...7d4f981. Read the comment docs.

[nsivabalan]

@hudi-bot run azure

[nsivabalan]

@hudi-bot run travis

[nsivabalan]

Few high level questions/comments.

• Now we have both clustering and compaction, I see that you have added clustering related code just after compaction where ever applicable. Is the higher priority for compaction intentional? or should we have clustering followed by compaction? or does it not matter at all.
• I came across a class named SchedulerConfGenerator. Don't we need to make any changes here for async clustering?

I am done reviewing source code. Yet to review tests.

[nsivabalan]

Here also we can think of something like AsyncServiceClient.
may be we can have a common method as below.

   public abstract void doAction(HoodieInstant instant) throw IOException;

same abstract class for both clustering and compaction.
Not too strong on this suggestion though. Let's see what others have to say.

[nsivabalan]

also, lets try to add java docs for all public methods. I do understand that AbstractCompactor does not have java docs. Its fine. atleast for the code we write, lets try to add java docs.

[codope]

same abstract class for both clustering and compaction.
Not too strong on this suggestion though. Let's see what others have to say.

@satishkotha what are your thoughts about this?

[codope]

@nsivabalan Thanks for reviewing the PR. I agree with your source code comments. There is scope for reusability. I will address them and update the PR. For the high level questions, my response is as below.

• Now we have both clustering and compaction, I see that you have added clustering related code just after compaction where ever applicable. Is the higher priority for compaction intentional? or should we have clustering followed by compaction? or does it not matter at all.

In case when both clustering and compaction are enabled then compaction will run just before clustering. The intention is that since currently compaction and clustering cannot run at the same time on the same file groups and clustering could take significant time, so let compaction thread start first. When clustering is scheduled for the filegroups under compaction it would be ignored and picked up in the subsequent run after compaction completes.

• I came across a class named SchedulerConfGenerator. Don't we need to make any changes here for async clustering?

We will need to make changes here if we create separate job pool for clustering and assign weights for different jobs. Unlike compaction, I did not feel the need for a separate job pool for clustering. By default, each pool gets equal share of resource but within each pool, jobs run in FIFO order.

[nsivabalan]

Yet to review tests. Will review in a day or two.

[nsivabalan]

out of curiosity. how come we have maxPendingCompaction/Clustering property defined as first class(top level) config for multiTableDeltaStreamer, but don't see the properties to enable/disable them. I assume those are fetched from property file for each source. So, why not fetch these properties also from the property file? I know this is not specific to clustering, but it was how compaction was defined.

[codope]

That's a valid point. I followed how we defined for compaction. We could have enable/disable configs as first class configs. Anything related to a particular feature/service could be passed in property file or as --hoodie-conf. This will avoid config bloating in delta streamer.

@bvaradar Was there any reason for doing it this way?

[nsivabalan]

@pratyakshsharma ^

[nsivabalan]

Tests coverage:

• Do we have tests where both compaction and clustering are async triggered? (both spark streaming and deltastreamer) ?

[nsivabalan]

@pratyakshsharma ^

[codope]

Tests coverage:

• Do we have tests where both compaction and clustering are async triggered? (both spark streaming and deltastreamer) ?

Added such test in TestHoodieDeltaStreamer and TestStructuredStreaming.

[codope]

@hudi-bot run azure

[codope]

@hudi-bot run azure

[zhangyue19921010]

Hi @codope Just want to know, is this Async clustering function can handle the following scenarios and losing no data:

There are 3 small file groups named fg1, fg2 and fg3 contained file slice1, file slice2 and file slices3 separately.

When async schedule start to make a cluster plan but not finished, there is an inflight or requested commit for fg1 which will create file slice 11 based on file slice1. In other words file slice11 is creating but not committed ---> I believe this scene is similar to multi writers.

What does this async clustering function will do?
Will this clustering plan contains file slice1? if contained, I think the new data in file slice11 will be lost.

Looking forward to your reply, thanks a lot.

[codope]

Hi @codope Just want to know, is this Async clustering function can handle the following scenarios and losing no data:

There are 3 small file groups named fg1, fg2 and fg3 contained file slice1, file slice2 and file slices3 separately.

When async schedule start to make a cluster plan but not finished, there is an inflight or requested commit for fg1 which will create file slice 11 based on file slice1. In other words file slice11 is creating but not committed ---> I believe this scene is similar to multi writers.

What does this async clustering function will do?
Will this clustering plan contains file slice1? if contained, I think the new data in file slice11 will be lost.

Looking forward to your reply, thanks a lot.

@zhangyue19921010 It will depend on what point of time during clustering planning file slice11 is created. If it is before the ClusteringPlanStrategy#getFileSlicesEligibleForClustering is invoked then clustering plan will not contain file slice1. So, just like multi writers there is a race condition here. However, while actually clustering, the default (and currently only) strategy is to reject updates. So, it will throw exception after seeing that there is an a filegroup with update (in this case fg1). This should get picked up in the next run of clustering.

[zhangyue19921010]

Hi @codope Thanks for your response. What if after ClusteringPlanStrategy#getFileSlicesEligibleForClustering file slice11 were created which means file slice1 was picked up for clustering. After this plan execute, maybe new data in slice11 was lost.