Skip to content

[HUDI-1632] Supports merge on read write mode for Flink writer #2593

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yanghua merged 1 commit into from
Mar 1, 2021
Merged

[HUDI-1632] Supports merge on read write mode for Flink writer #2593

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 32 additions & 16 deletions hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,38 +74,38 @@ public class HoodieAppendHandle<T extends HoodieRecordPayload, I, K, O> extends
// This acts as the sequenceID for records written
private static final AtomicLong RECORD_COUNTER = new AtomicLong(1);

private final String fileId;
protected final String fileId;
// Buffer for holding records in memory before they are flushed to disk
private final List<IndexedRecord> recordList = new ArrayList<>();
// Buffer for holding records (to be deleted) in memory before they are flushed to disk
private final List<HoodieKey> keysToDelete = new ArrayList<>();
// Incoming records to be written to logs.
private final Iterator<HoodieRecord<T>> recordItr;
protected Iterator<HoodieRecord<T>> recordItr;
// Writer to log into the file group's latest slice.
private Writer writer;
protected Writer writer;

private final List<WriteStatus> statuses;
protected final List<WriteStatus> statuses;
// Total number of records written during an append
private long recordsWritten = 0;
protected long recordsWritten = 0;
// Total number of records deleted during an append
private long recordsDeleted = 0;
protected long recordsDeleted = 0;
// Total number of records updated during an append
private long updatedRecordsWritten = 0;
protected long updatedRecordsWritten = 0;
// Total number of new records inserted into the delta file
private long insertRecordsWritten = 0;
protected long insertRecordsWritten = 0;

// Average record size for a HoodieRecord. This size is updated at the end of every log block flushed to disk
private long averageRecordSize = 0;
// Flag used to initialize some metadata
private boolean doInit = true;
// Total number of bytes written during this append phase (an estimation)
private long estimatedNumberOfBytesWritten;
protected long estimatedNumberOfBytesWritten;
// Number of records that must be written to meet the max block size for a log block
private int numberOfRecords = 0;
// Max block size to limit to for a log block
private final int maxBlockSize = config.getLogFileDataBlockMaxSize();
// Header metadata for a log block
private final Map<HeaderMetadataType, String> header = new HashMap<>();
protected final Map<HeaderMetadataType, String> header = new HashMap<>();
private SizeEstimator<HoodieRecord> sizeEstimator;

public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
Expand Down Expand Up @@ -178,6 +178,14 @@ private void init(HoodieRecord record) {
}
}

/**
* Returns whether the hoodie record is an UPDATE.
*/
protected boolean isUpdateRecord(HoodieRecord<T> hoodieRecord) {
// If currentLocation is present, then this is an update
return hoodieRecord.getCurrentLocation() != null;

@yanghua yanghua Feb 26, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In previous PRs, you used instantTime of HoodieRecordLocation to judge update operation, right? So, let keep the same?

@danny0405 danny0405 Feb 26, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the default logic for Hoodie, we should keep it, the FlinkAppendHandle already overrides the behavior.

@yanghua yanghua Feb 26, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it's in the common package, my problem.

}

private Option<IndexedRecord> getIndexedRecord(HoodieRecord<T> hoodieRecord) {
Option<Map<String, String>> recordMetadata = hoodieRecord.getData().getMetadata();
try {
Expand All @@ -190,8 +198,7 @@ private Option<IndexedRecord> getIndexedRecord(HoodieRecord<T> hoodieRecord) {
HoodieAvroUtils.addHoodieKeyToRecord((GenericRecord) avroRecord.get(), hoodieRecord.getRecordKey(),
hoodieRecord.getPartitionPath(), fileId);
HoodieAvroUtils.addCommitMetadataToRecord((GenericRecord) avroRecord.get(), instantTime, seqId);
// If currentLocation is present, then this is an update
if (hoodieRecord.getCurrentLocation() != null) {
if (isUpdateRecord(hoodieRecord)) {
updatedRecordsWritten++;
} else {
insertRecordsWritten++;
Expand Down Expand Up @@ -324,7 +331,7 @@ public void doAppend() {
estimatedNumberOfBytesWritten += averageRecordSize * numberOfRecords;
}

private void appendDataAndDeleteBlocks(Map<HeaderMetadataType, String> header) {
protected void appendDataAndDeleteBlocks(Map<HeaderMetadataType, String> header) {
try {
header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, instantTime);
header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, writerSchemaWithMetafields.toString());
Expand Down Expand Up @@ -412,6 +419,13 @@ private Writer createLogWriter(Option<FileSlice> fileSlice, String baseCommitTim
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).build();
}

/**
* Whether there is need to update the record location.
*/
protected boolean needsUpdateLocation() {
return true;
}

private void writeToBuffer(HoodieRecord<T> record) {
if (!partitionPath.equals(record.getPartitionPath())) {
HoodieUpsertException failureEx = new HoodieUpsertException("mismatched partition path, record partition: "
Expand All @@ -421,9 +435,11 @@ private void writeToBuffer(HoodieRecord<T> record) {
}

// update the new location of the record, so we know where to find it next
record.unseal();
record.setNewLocation(new HoodieRecordLocation(instantTime, fileId));
record.seal();
if (needsUpdateLocation()) {
record.unseal();
record.setNewLocation(new HoodieRecordLocation(instantTime, fileId));
record.seal();
}
Option<IndexedRecord> indexedRecord = getIndexedRecord(record);
if (indexedRecord.isPresent()) {
recordList.add(indexedRecord.get());
Expand Down
15 changes: 12 additions & 3 deletions hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,13 @@ protected void initializeIncomingRecordsMap() {
}
}

/**
* Whether there is need to update the record location.
*/
boolean needsUpdateLocation() {
return true;
}

/**
* Load the new incoming records in a map and return partitionPath.
*/
Expand All @@ -206,9 +213,11 @@ protected void init(String fileId, Iterator<HoodieRecord<T>> newRecordsItr) {
while (newRecordsItr.hasNext()) {
HoodieRecord<T> record = newRecordsItr.next();
// update the new location of the record, so we know where to find it next
record.unseal();
record.setNewLocation(new HoodieRecordLocation(instantTime, fileId));
record.seal();
if (needsUpdateLocation()) {
record.unseal();
record.setNewLocation(new HoodieRecordLocation(instantTime, fileId));
record.seal();
}
// NOTE: Once Records are added to map (spillable-map), DO NOT change it as they won't persist
keyToNewRecords.put(record.getRecordKey(), record);
}
Expand Down
121 changes: 101 additions & 20 deletions ...client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,11 @@
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieCommitException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.index.FlinkHoodieIndex;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.io.FlinkAppendHandle;
import org.apache.hudi.io.FlinkCreateHandle;
import org.apache.hudi.io.FlinkMergeHandle;
import org.apache.hudi.io.HoodieWriteHandle;
Expand All @@ -48,14 +50,19 @@
import org.apache.hudi.table.HoodieFlinkTable;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.hudi.table.action.compact.FlinkCompactHelpers;
import org.apache.hudi.table.upgrade.FlinkUpgradeDowngrade;

import com.codahale.metrics.Timer;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.ParseException;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
Expand All @@ -64,6 +71,8 @@
public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
AbstractHoodieWriteClient<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> {

private static final Logger LOG = LoggerFactory.getLogger(HoodieFlinkWriteClient.class);

/**
* FileID to write handle mapping in order to record the write handles for each file group,
* so that we can append the mini-batch data buffer incrementally.
Expand Down Expand Up @@ -127,21 +136,9 @@ public List<WriteStatus> upsert(List<HoodieRecord<T>> records, String instantTim
table.validateUpsertSchema();
preWrite(instantTime, WriteOperationType.UPSERT);
final HoodieRecord<T> record = records.get(0);
final HoodieRecordLocation loc = record.getCurrentLocation();
final String fileID = loc.getFileId();
final boolean isInsert = loc.getInstantTime().equals("I");
final HoodieWriteHandle<?, ?, ?, ?> writeHandle;
if (bucketToHandles.containsKey(fileID)) {
writeHandle = bucketToHandles.get(fileID);
} else {
// create the write handle if not exists
writeHandle = isInsert
? new FlinkCreateHandle<>(getConfig(), instantTime, table, record.getPartitionPath(),
fileID, table.getTaskContextSupplier())
: new FlinkMergeHandle<>(getConfig(), instantTime, table, records.listIterator(), record.getPartitionPath(),
fileID, table.getTaskContextSupplier());
bucketToHandles.put(fileID, writeHandle);
}
final boolean isDelta = table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ);
final HoodieWriteHandle<?, ?, ?, ?> writeHandle = getOrCreateWriteHandle(record, isDelta, getConfig(),
instantTime, table, record.getPartitionPath(), records.listIterator());
HoodieWriteMetadata<List<WriteStatus>> result = ((HoodieFlinkTable<T>) table).upsert(context, writeHandle, instantTime, records);
if (result.getIndexLookupDuration().isPresent()) {
metrics.updateIndexMetrics(LOOKUP_STR, result.getIndexLookupDuration().get().toMillis());
Expand All @@ -160,7 +157,12 @@ public List<WriteStatus> insert(List<HoodieRecord<T>> records, String instantTim
getTableAndInitCtx(WriteOperationType.INSERT, instantTime);
table.validateUpsertSchema();
preWrite(instantTime, WriteOperationType.INSERT);
HoodieWriteMetadata<List<WriteStatus>> result = table.insert(context, instantTime, records);
// create the write handle if not exists
final HoodieRecord<T> record = records.get(0);
final boolean isDelta = table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ);
final HoodieWriteHandle<?, ?, ?, ?> writeHandle = getOrCreateWriteHandle(record, isDelta, getConfig(),
instantTime, table, record.getPartitionPath(), records.listIterator());
HoodieWriteMetadata<List<WriteStatus>> result = ((HoodieFlinkTable<T>) table).insert(context, writeHandle, instantTime, records);
if (result.getIndexLookupDuration().isPresent()) {
metrics.updateIndexMetrics(LOOKUP_STR, result.getIndexLookupDuration().get().toMillis());
}
Expand Down Expand Up @@ -207,13 +209,40 @@ protected List<WriteStatus> postWrite(HoodieWriteMetadata<List<WriteStatus>> res
}

@Override
public void commitCompaction(String compactionInstantTime, List<WriteStatus> writeStatuses, Option<Map<String, String>> extraMetadata) throws IOException {
throw new HoodieNotSupportedException("Compaction is not supported yet");
public void commitCompaction(
String compactionInstantTime,
List<WriteStatus> writeStatuses,
Option<Map<String, String>> extraMetadata) throws IOException {
HoodieFlinkTable<T> table = HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) context);
HoodieCommitMetadata metadata = FlinkCompactHelpers.newInstance().createCompactionMetadata(
table, compactionInstantTime, writeStatuses, config.getSchema());
extraMetadata.ifPresent(m -> m.forEach(metadata::addMetadata));
completeCompaction(metadata, writeStatuses, table, compactionInstantTime);
}

@simonsssu simonsssu Feb 24, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems it just contains commitCompaction and planScheduler, where is the real compact logic ? I see there exists a CompactFunction and CompactCommitSink , but I don't find where to call these.

@danny0405 danny0405 Feb 24, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can take StreamWriteITCase#testMergeOnReadWriteWithCompaction for an example.

@simonsssu simonsssu Feb 24, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand the planscheduler to generate plan, and then commitCompaction will read the plan to see if it should commit, but I'm confused to connect all these together, when to trigger the real compact ? if we use CompactFunction , how to understand the async here ?

@danny0405 danny0405 Feb 24, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The StreamWriteOperatorCoordinator.checkpointComplete is the entry point to schedule a compaction, but because there is a pluggable strategy there, a schedule may generate a null compaction plan (E.G. no compaction), we say the compaction is async.

@simonsssu simonsssu Feb 24, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for explanation.

@Override
protected void completeCompaction(HoodieCommitMetadata metadata, List<WriteStatus> writeStatuses, HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table, String compactionCommitTime) {
throw new HoodieNotSupportedException("Compaction is not supported yet");
public void completeCompaction(
HoodieCommitMetadata metadata,
List<WriteStatus> writeStatuses,
HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table,
String compactionCommitTime) {
this.context.setJobStatus(this.getClass().getSimpleName(), "Collect compaction write status and commit compaction");
List<HoodieWriteStat> writeStats = writeStatuses.stream().map(WriteStatus::getStat).collect(Collectors.toList());
finalizeWrite(table, compactionCommitTime, writeStats);
LOG.info("Committing Compaction {} finished with result {}.", compactionCommitTime, metadata);
FlinkCompactHelpers.newInstance().completeInflightCompaction(table, compactionCommitTime, metadata);

if (compactionTimer != null) {
long durationInMs = metrics.getDurationInMs(compactionTimer.stop());
try {
metrics.updateCommitMetrics(HoodieActiveTimeline.COMMIT_FORMATTER.parse(compactionCommitTime).getTime(),
durationInMs, metadata, HoodieActiveTimeline.COMPACTION_ACTION);
} catch (ParseException e) {
throw new HoodieCommitException("Commit time is not of valid format. Failed to commit compaction "
+ config.getBasePath() + " at time " + compactionCommitTime, e);
}
}
LOG.info("Compacted successfully on commit " + compactionCommitTime);
}

@Override
Expand Down Expand Up @@ -244,6 +273,46 @@ public void cleanHandles() {
this.bucketToHandles.clear();
}

/**
* Get or create a new write handle in order to reuse the file handles.
*
* @param record The first record in the bucket
* @param isDelta Whether the table is in MOR mode
* @param config Write config
* @param instantTime The instant time
* @param table The table
* @param partitionPath Partition path
* @param recordItr Record iterator
* @return Existing write handle or create a new one
*/
private HoodieWriteHandle<?, ?, ?, ?> getOrCreateWriteHandle(
HoodieRecord<T> record,
boolean isDelta,
HoodieWriteConfig config,
String instantTime,
HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table,
String partitionPath,
Iterator<HoodieRecord<T>> recordItr) {
final HoodieRecordLocation loc = record.getCurrentLocation();
final String fileID = loc.getFileId();
if (bucketToHandles.containsKey(fileID)) {
return bucketToHandles.get(fileID);
}
final HoodieWriteHandle<?, ?, ?, ?> writeHandle;
if (isDelta) {
Comment thread
yanghua marked this conversation as resolved.
Outdated
writeHandle = new FlinkAppendHandle<>(config, instantTime, table, partitionPath, fileID, recordItr,
table.getTaskContextSupplier());
} else if (loc.getInstantTime().equals("I")) {
writeHandle = new FlinkCreateHandle<>(config, instantTime, table, partitionPath,
fileID, table.getTaskContextSupplier());
} else {
writeHandle = new FlinkMergeHandle<>(config, instantTime, table, recordItr, partitionPath,
fileID, table.getTaskContextSupplier());
}
this.bucketToHandles.put(fileID, writeHandle);
return writeHandle;
}

private HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> getTableAndInitCtx(HoodieTableMetaClient metaClient, WriteOperationType operationType) {
if (operationType == WriteOperationType.DELETE) {
setWriteSchemaForDeletes(metaClient);
Expand Down Expand Up @@ -305,4 +374,16 @@ public void transitionRequestedToInflight(String tableType, String inFlightInsta
activeTimeline.transitionRequestedToInflight(requested, Option.empty(),
config.shouldAllowMultiWriteOnSameInstant());
}

public void rollbackInflightCompaction(HoodieInstant inflightInstant) {
HoodieFlinkTable<T> table = HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) context);
HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
rollbackInflightCompaction(inflightInstant, table);
}
}

public HoodieFlinkTable<T> getHoodieTable() {
return HoodieFlinkTable.create(config, (HoodieFlinkEngineContext) context);
}
}
10 changes: 0 additions & 10 deletions .../hudi-flink-client/src/main/java/org/apache/hudi/index/state/FlinkInMemoryStateIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,16 +23,12 @@
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieIndexException;
import org.apache.hudi.index.FlinkHoodieIndex;
import org.apache.hudi.table.HoodieTable;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

Expand All @@ -46,15 +42,9 @@
public class FlinkInMemoryStateIndex<T extends HoodieRecordPayload> extends FlinkHoodieIndex<T> {

private static final Logger LOG = LogManager.getLogger(FlinkInMemoryStateIndex.class);
private MapState<HoodieKey, HoodieRecordLocation> mapState;

public FlinkInMemoryStateIndex(HoodieFlinkEngineContext context, HoodieWriteConfig config) {
super(config);
if (context.getRuntimeContext() != null) {
MapStateDescriptor<HoodieKey, HoodieRecordLocation> indexStateDesc =
new MapStateDescriptor<>("indexState", TypeInformation.of(HoodieKey.class), TypeInformation.of(HoodieRecordLocation.class));
mapState = context.getRuntimeContext().getMapState(indexStateDesc);
}
}

@Override
Expand Down
6 changes: 3 additions & 3 deletions .../hudi/io/ExplicitCreateHandleFactory.java → ...e/hudi/io/ExplicitWriteHandleFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,11 @@
/**
* Create handle factory for Flink writer, use the specified write handle directly.
*/
public class ExplicitCreateHandleFactory<T extends HoodieRecordPayload, I, K, O>
extends CreateHandleFactory<T, I, K, O> {
public class ExplicitWriteHandleFactory<T extends HoodieRecordPayload, I, K, O>
extends WriteHandleFactory<T, I, K, O> {
private HoodieWriteHandle<T, I, K, O> writeHandle;

public ExplicitCreateHandleFactory(HoodieWriteHandle<T, I, K, O> writeHandle) {
public ExplicitWriteHandleFactory(HoodieWriteHandle<T, I, K, O> writeHandle) {
this.writeHandle = writeHandle;
}

Expand Down
Loading