[MINOR] Improve code readability by passing in the fileComparisonsRDD in bloom index

[danny0405]

What is the purpose of the pull request

The option 'hoodie.bloom.index.prune.by.ranges' is default true, so it
expects to be an improvement.

Brief change log

• Some refactoring to SparkHoodieBloomIndex

Verify this pull request

Existing tests

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.

[garyli1019]

@danny0405 Thanks for your contribution, good catch! The first compute probably being cached, but this change definitely looks better.
Waiting for another reviewer to double-check. @nsivabalan mind taking a look?

[yanghua]

+1 too, just left one minor comment.

[yanghua]

Maybe we can move fileGroupToComparisons into this line or break the hoodieTable to the next line. It would be better?

[danny0405]

Thanks, agree ~

[vinothchandar]

Thanks for the PR @danny0405 !

This leaves the code easier to read. but unsure of the intended effects. I have actually thought more about how to avoid this. and it seems hard. we do need some evaluation of how much pruning is going to happen for us to size the parallelism and since that is very workload specific, it does need an evaluation.

best idea I could come up with, was to see if we can do some sampling say 10% of the records to estimate.

Something like below.

/ we will just try exploding the input and then count to determine comparisons
      // FIX(vc): Only do sampling here and extrapolate?
      fileToComparisons = explodeRecordRDDWithFileComparisons(partitionToFileInfo, partitionRecordKeyPairRDD.sample(0.1))
          .mapToPair(t -> t).countByKey();

[vinothchandar]

nit: wondering how checkstyle is happy with the indentation here. :)

[danny0405]

Thanks, i just notice that Hoodie use the whitespace for indentation.

[vinothchandar]

do you actually see from the Spark UI that its not computed twice? I ask because, fileComparisonsRDD is not cached and thus even though this is declared only once, during runtime, Spark will lazily recompute fileComparisonsRDD once for each method that uses it.

[danny0405]

I didn't check the Spark UI yet, just a simple analyze the process of data writing. For each batch of records to write, the SparkHoodieBloomIndex.lookupIndex was expected to be invoked once so the fileComparisonsRDD should only be evaluated only once, is there other invocation for SparkHoodieBloomIndex.lookupIndex ? Maybe i missed something.

[vinothchandar]

So Spark does lazy evaluation of an RDD. If the RDD is not persisted to disk/cache, it simply recomputes it. In this case, fileComparisonsRDD would be recomputed twice during runtime. the method explodeRecordRDDWithFileComparisons() will only be called once, but it does nothing in practice except "define" the RDD it returns.

In contrast, if you notice this line of code at the start of tagLocation

// Step 0: cache the input record RDD
    if (config.getBloomIndexUseCaching()) {
      recordRDD.persist(SparkMemoryUtils.getBloomIndexInputStorageLevel(config.getProps()));
    }

This caches the incoming recordRDD and thus however many times this RDD is used in the indexing DAG, it will not go to the previous stage. if we did not have the .persist() in here, then everytime an RDD derived off this recordRDD is needed for a Spark action, it will keep reading from source and compute the recordRDD again.

Apologies, if you knew all this already. :) and I am failing to see how this wont happen. but at-least you get my concern with this explanation.

[danny0405]

Thanks for the explanation, the recordRDD may be persisted, but the computation in explodeRecordRDDWithFileComparisons still need to do 2 times, right ? Sorry, i'm not that familiar with the RDD thing.

[vinothchandar]

Correct. That's what i think will happen

[danny0405]

Thanks for the PR @danny0405 !

This leaves the code easier to read. but unsure of the intended effects. I have actually thought more about how to avoid this. and it seems hard. we do need some evaluation of how much pruning is going to happen for us to size the parallelism and since that is very workload specific, it does need an evaluation.

best idea I could come up with, was to see if we can do some sampling say 10% of the records to estimate.

Something like below.

/ we will just try exploding the input and then count to determine comparisons
      // FIX(vc): Only do sampling here and extrapolate?
      fileToComparisons = explodeRecordRDDWithFileComparisons(partitionToFileInfo, partitionRecordKeyPairRDD.sample(0.1))
          .mapToPair(t -> t).countByKey();

Thanks so much for the review @vinothchandar ~ Can we really do a sampling here ? Then how can we ensure the FileId -> HoodieKey candidates are complete so that the final built Index is accurate. (e.g. No UPDATE records are tagged as INSERT).

[codecov-io]

Codecov Report

Merging #2319 (622c3c7) into master (bd9ccec) will decrease coverage by 41.88%.
The diff coverage is n/a.

@@              Coverage Diff              @@
##             master    #2319       +/-   ##
=============================================
- Coverage     52.29%   10.40%   -41.89%     
+ Complexity     2630       48     -2582     
=============================================
  Files           329       51      -278     
  Lines         14743     1787    -12956     
  Branches       1484      213     -1271     
=============================================
- Hits           7710      186     -7524     
+ Misses         6419     1588     -4831     
+ Partials        614       13      -601     

Flag	Coverage Δ	Complexity Δ
hudicli	?	?
hudiclient	?	?
hudicommon	?	?
hudihadoopmr	?	?
huditimelineservice	?	?
hudiutilities	10.40% <ø> (-59.71%)	0.00 <ø> (ø)

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

Impacted Files	Coverage Δ	Complexity Δ
...va/org/apache/hudi/utilities/IdentitySplitter.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-2.00%)
...va/org/apache/hudi/utilities/schema/SchemaSet.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-3.00%)
...a/org/apache/hudi/utilities/sources/RowSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-4.00%)
.../org/apache/hudi/utilities/sources/AvroSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-1.00%)
.../org/apache/hudi/utilities/sources/JsonSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-1.00%)
...rg/apache/hudi/utilities/sources/CsvDFSSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-10.00%)
...g/apache/hudi/utilities/sources/JsonDFSSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-4.00%)
...apache/hudi/utilities/sources/JsonKafkaSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-6.00%)
...pache/hudi/utilities/sources/ParquetDFSSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-5.00%)
...lities/schema/SchemaProviderWithPostProcessor.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-4.00%)
... and 306 more

[vinothchandar]

Can we really do a sampling here ?

yes, there is a trade-off here. We have to rely on the fact that the previous stage does a reduceByKey so the data would be distributed evenly across the RDD partitions :). So, extrapolation could be okay, for purposes of computeComparisonsPerFileGroup(). The other usage, where we actually do the range pruning, we cannot do sampling.

[danny0405]

The other usage, where we actually do the range pruning, we cannot do sampling

Thanks, got your idea ~

[vinothchandar]

@danny0405 I ll take another pass and merge this PR. This should not have any side effects so should qualify for a MINOR label.

IIRC there are some JIRAs with the index/performance component around some followup ideas we discussed here

[nsivabalan]

yeah, same was already brought up before and we didn't proceed since it needed some perf analysis.
this was the rational:
even though explodeRecordRDDWithFileComparisons(...) is called in two places, in the first place we just do count by key which may not shuffle any actual data, where as the 2nd call could incur shuffling. Hence the pattern of usage differs. More info can be found in the attached PR.

[vinothchandar]

Merging.