[HUDI-6344] Flink MDT bulk_insert for initial commit #8914
Conversation
|
@hudi-bot run azure |
|
All the Flink related tests are passed: https://dev.azure.com/apache-hudi-ci-org/apache-hudi-ci/_build/results?buildId=17702&view=results |
| .collect(Collectors.groupingBy(r -> r.getCurrentLocation().getFileId())); | ||
| return preppedRecordsByFileId.values().stream().parallel().map(records -> { | ||
| HoodieWriteMetadata<List<WriteStatus>> result; | ||
| records.get(0).getCurrentLocation().setInstantTime("I"); |
There was a problem hiding this comment.
Why setting instantTime to I? Should it not be instantTime passed to the method?
There was a problem hiding this comment.
Flink relies on this flag to distinguish which write handle to use. A little hacky but it's the minimum change right now.
| ? writeClient.upsertPreppedRecords(preppedRecordList, instantTime) | ||
| : Collections.emptyList(); | ||
| List<WriteStatus> statuses = isInitializing | ||
| ? writeClient.bulkInsertPreppedRecords(preppedRecordList, instantTime, Option.empty()) |
There was a problem hiding this comment.
major reason to use bulkInsert is that, we use a custom partitioner based on file group and so the spark tasks will be such that, each spark task will get records pertaining to one file group of interest.
we can try to incorporate that as well. esply with RLI, record mapping to file groups is based on hash. So, we can't have diff set of records routed to one spark task.
There was a problem hiding this comment.
Flink does not support flexible partitioner like Spark do, but we can still get some benefits because writing Hfiles directly for initilization of MDT is more efficient. For example, when user enable the MDT for a existing table and there are plenty of metadata records to bootstrap with.
There was a problem hiding this comment.
records to file group mapping is deterministic and we can have only one file written per file group. for eg, if we instanttiate col stats with 4 file groups, we should spin up 4 spark tasks and each spark task should get records pertaining to the file group of interest (remember records are mapped to file group based on hashing). So, if one spark task gets records for all file groups, then we might end up w/ n*m files (where n is no of spark tasks and m is number of file groups) which may not work. we need only m files created and m spark tasks should spin up where each spark tasks writes to just 1 file group.
hope that makes sense.
There was a problem hiding this comment.
What the spark partitioner does is to repartition the records by the file group (index) to avoid concurrent write from different tasks into one file group. Flink already does that, even though it is parallelized in single JVM process.
There was a problem hiding this comment.
if its already taken care of, then we are good
4 participants
Change Logs
Fix the bulk_insert for Flink MDT initialization after #8684 .
Impact
none
Risk level (write none, low medium or high below)
none
Documentation Update
Describe any necessary documentation update if there is any new feature, config, or user-facing change
ticket number here and follow the instruction to make
changes to the website.
Contributor's checklist