[HUDI-1598] Write as minor batches during one checkpoint interval for… #2553
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.hudi.io; | ||
|
|
||
| import org.apache.hudi.client.WriteStatus; | ||
| import org.apache.hudi.common.engine.TaskContextSupplier; | ||
| import org.apache.hudi.common.fs.FSUtils; | ||
| import org.apache.hudi.common.model.HoodieRecord; | ||
| import org.apache.hudi.common.model.HoodieRecordPayload; | ||
| import org.apache.hudi.common.model.HoodieWriteStat; | ||
| import org.apache.hudi.common.util.HoodieTimer; | ||
| import org.apache.hudi.common.util.collection.Pair; | ||
| import org.apache.hudi.config.HoodieWriteConfig; | ||
| import org.apache.hudi.exception.HoodieInsertException; | ||
| import org.apache.hudi.table.HoodieTable; | ||
|
|
||
| import org.apache.avro.Schema; | ||
| import org.apache.hadoop.fs.Path; | ||
| import org.apache.log4j.LogManager; | ||
| import org.apache.log4j.Logger; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * A {@link HoodieCreateHandle} that supports create write incrementally(mini-batches). | ||
| * | ||
| * <p>For the first mini-batch, it initialize and set up the next file path to write, | ||
| * but does not close the file writer until all the mini-batches write finish. Each mini-batch | ||
| * data are appended to the same file. | ||
| * | ||
| * @param <T> Payload type | ||
| * @param <I> Input type | ||
| * @param <K> Key type | ||
| * @param <O> Output type | ||
| */ | ||
| public class FlinkCreateHandle<T extends HoodieRecordPayload, I, K, O> | ||
| extends HoodieCreateHandle<T, I, K, O> implements MiniBatchHandle { | ||
|
|
||
| private static final Logger LOG = LogManager.getLogger(FlinkCreateHandle.class); | ||
| private long lastFileSize = 0L; | ||
|
|
||
| public FlinkCreateHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable, | ||
| String partitionPath, String fileId, TaskContextSupplier taskContextSupplier) { | ||
| this(config, instantTime, hoodieTable, partitionPath, fileId, getWriterSchemaIncludingAndExcludingMetadataPair(config), | ||
| taskContextSupplier); | ||
| } | ||
|
|
||
| public FlinkCreateHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable, | ||
| String partitionPath, String fileId, Pair<Schema, Schema> writerSchemaIncludingAndExcludingMetadataPair, | ||
| TaskContextSupplier taskContextSupplier) { | ||
| super(config, instantTime, hoodieTable, partitionPath, fileId, writerSchemaIncludingAndExcludingMetadataPair, | ||
| taskContextSupplier); | ||
| } | ||
|
|
||
| /** | ||
| * Called by the compactor code path. | ||
| */ | ||
| public FlinkCreateHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable, | ||
| String partitionPath, String fileId, Map<String, HoodieRecord<T>> recordMap, | ||
| TaskContextSupplier taskContextSupplier) { | ||
| super(config, instantTime, hoodieTable, partitionPath, fileId, recordMap, taskContextSupplier); | ||
| } | ||
|
|
||
| /** | ||
| * Get the incremental write status. In mini-batch write mode, | ||
| * this handle would be reused for a checkpoint bucket(the bucket is appended as mini-batches), | ||
|
||
| * thus, after a mini-batch append finish, we do not close the underneath writer but return | ||
| * the incremental WriteStatus instead. | ||
| * | ||
| * @return the incremental write status | ||
| */ | ||
| private WriteStatus getIncrementalWriteStatus() { | ||
| try { | ||
| long fileSizeInBytes = FSUtils.getFileSize(fs, path); | ||
| setUpWriteStatus(fileSizeInBytes); | ||
| // reset the write status | ||
| recordsWritten = 0; | ||
| recordsDeleted = 0; | ||
| insertRecordsWritten = 0; | ||
| this.lastFileSize = fileSizeInBytes; | ||
| writeStatus.setTotalErrorRecords(0); | ||
| return writeStatus; | ||
| } catch (IOException e) { | ||
| throw new HoodieInsertException("Failed to close the Insert Handle for path " + path, e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Set up the write status. | ||
| * | ||
| * @param fileSizeInBytes File size in bytes | ||
| * @throws IOException if error occurs | ||
| */ | ||
| private void setUpWriteStatus(long fileSizeInBytes) throws IOException { | ||
| HoodieWriteStat stat = new HoodieWriteStat(); | ||
| stat.setPartitionPath(writeStatus.getPartitionPath()); | ||
| stat.setNumWrites(recordsWritten); | ||
| stat.setNumDeletes(recordsDeleted); | ||
| stat.setNumInserts(insertRecordsWritten); | ||
| stat.setPrevCommit(HoodieWriteStat.NULL_COMMIT); | ||
| stat.setFileId(writeStatus.getFileId()); | ||
| stat.setPath(new Path(config.getBasePath()), path); | ||
| stat.setTotalWriteBytes(fileSizeInBytes - lastFileSize); | ||
| stat.setFileSizeInBytes(fileSizeInBytes); | ||
| stat.setTotalWriteErrors(writeStatus.getTotalErrorRecords()); | ||
| HoodieWriteStat.RuntimeStats runtimeStats = new HoodieWriteStat.RuntimeStats(); | ||
| runtimeStats.setTotalCreateTime(timer.endTimer()); | ||
| stat.setRuntimeStats(runtimeStats); | ||
| timer = new HoodieTimer().startTimer(); | ||
| writeStatus.setStat(stat); | ||
| } | ||
|
|
||
| public void finishWrite() { | ||
| LOG.info("Closing the file " + writeStatus.getFileId() + " as we are done with all the records " + recordsWritten); | ||
| try { | ||
| fileWriter.close(); | ||
| } catch (IOException e) { | ||
| throw new HoodieInsertException("Failed to close the Insert Handle for path " + path, e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Performs actions to durably, persist the current changes and returns a WriteStatus object. | ||
| */ | ||
| @Override | ||
| public List<WriteStatus> close() { | ||
| return Collections.singletonList(getIncrementalWriteStatus()); | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why use first record' bucket type? if first record is
Iand the second one isU, th code will useFlinkCreateHandleto process all recordsThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, that is the logic we want, because we try to merge records into small buckets.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So we also need to filter dupicate records on the query side ?