Skip to content

HUDI-4044 When reading data from flink-hudi to external storage, the … #5516

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
danny0405 merged 5 commits into from
May 10, 2022
Merged

HUDI-4044 When reading data from flink-hudi to external storage, the … #5516

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
75ef37e
HUDI-4044 When reading data from flink-hudi to external storage, the …
May 6, 2022
f88f4a6
HUDI-4044 When reading data from flink-hudi to external storage, the …
May 9, 2022
6b98aa1
HUDI-4044 When reading data from flink-hudi to external storage, the …
May 9, 2022
0045f11
HUDI-4044 When reading data from flink-hudi to external storage, the …
May 9, 2022
0cfbdfc
HUDI-4044 When reading data from flink-hudi to external storage, the …
May 9, 2022
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
2 changes: 1 addition & 1 deletion ...nk-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,7 @@ public Result inputSplits(
String basePath = fileSlice.getBaseFile().map(BaseFile::getPath).orElse(null);
return new MergeOnReadInputSplit(cnt.getAndAdd(1),
basePath, logPaths, endInstant,
metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType, instantRange);
metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType, instantRange, fileSlice.getFileId());
}).collect(Collectors.toList()))
Copy link
Contributor

@danny0405 danny0405 May 9, 2022

Choose a reason for hiding this comment

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

instantRange,fileSlice.getFileId() -> instantRange, fileSlice.getFileId()

.flatMap(Collection::stream)
.collect(Collectors.toList());
Expand Down
6 changes: 4 additions & 2 deletions hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.hudi.table;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.common.model.BaseFile;
import org.apache.hudi.common.model.HoodieLogFile;
Expand Down Expand Up @@ -181,7 +182,8 @@ public DataStream<RowData> produceDataStream(StreamExecutionEnvironment execEnv)
OneInputStreamOperatorFactory<MergeOnReadInputSplit, RowData> factory = StreamReadOperator.factory((MergeOnReadInputFormat) inputFormat);
SingleOutputStreamOperator<RowData> source = execEnv.addSource(monitoringFunction, getSourceOperatorName("split_monitor"))
.setParallelism(1)
.transform("split_reader", typeInfo, factory)
.keyBy((KeySelector<MergeOnReadInputSplit, String>) mos -> mos.getFileId())
.transform("split_reader", typeInfo, factory)
Copy link
Contributor

@danny0405 danny0405 May 9, 2022

Choose a reason for hiding this comment

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

Is the explicit cast necessary ?

Copy link
Contributor

@danny0405 danny0405 May 9, 2022

Choose a reason for hiding this comment

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

Can the explicit KeySelector be removed ?

.setParallelism(conf.getInteger(FlinkOptions.READ_TASKS));
return new DataStreamSource<>(source);
} else {
Expand Down Expand Up @@ -316,7 +318,7 @@ private List<MergeOnReadInputSplit> buildFileIndex() {
.map(logFile -> logFile.getPath().toString())
.collect(Collectors.toList()));
return new MergeOnReadInputSplit(cnt.getAndAdd(1), basePath, logPaths, latestCommit,
metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType, null);
metaClient.getBasePath(), maxCompactionMemoryInBytes, mergeType, null, fileSlice.getFileId());
}).collect(Collectors.toList()))
Copy link
Contributor

@danny0405 danny0405 May 9, 2022

Choose a reason for hiding this comment

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

null,fileSlice.getFileId() -> null, fileSlice.getFileId()

.flatMap(Collection::stream)
.collect(Collectors.toList());
Expand Down
34 changes: 23 additions & 11 deletions ...urce/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,11 @@

package org.apache.hudi.table.format.mor;

import org.apache.flink.core.io.InputSplit;
import org.apache.hudi.common.table.log.InstantRange;
import org.apache.hudi.common.util.Option;

import org.apache.flink.core.io.InputSplit;

Copy link
Contributor

@danny0405 danny0405 May 9, 2022

Choose a reason for hiding this comment

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

The changes for the import is not necessary.

import javax.annotation.Nullable;

import java.util.List;

/**
Expand All @@ -43,20 +41,25 @@ public class MergeOnReadInputSplit implements InputSplit {
private final long maxCompactionMemoryInBytes;
private final String mergeType;
private final Option<InstantRange> instantRange;
private String fileId;


// for streaming reader to record the consumed offset,
// which is the start of next round reading.
private long consumed = NUM_NO_CONSUMPTION;



public MergeOnReadInputSplit(
int splitNum,
@Nullable String basePath,
Option<List<String>> logPaths,
String latestCommit,
String tablePath,
long maxCompactionMemoryInBytes,
String mergeType,
@Nullable InstantRange instantRange) {
int splitNum,
@Nullable String basePath,
Option<List<String>> logPaths,
String latestCommit,
Copy link
Contributor

@danny0405 danny0405 May 9, 2022

Choose a reason for hiding this comment

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

Fix the indentation

String tablePath,
long maxCompactionMemoryInBytes,
String mergeType,
@Nullable InstantRange instantRange,
String fileId) {
this.splitNum = splitNum;
this.basePath = Option.ofNullable(basePath);
this.logPaths = logPaths;
Expand All @@ -65,6 +68,15 @@ public MergeOnReadInputSplit(
this.maxCompactionMemoryInBytes = maxCompactionMemoryInBytes;
this.mergeType = mergeType;
this.instantRange = Option.ofNullable(instantRange);
this.fileId = fileId;
}

public String getFileId() {
return fileId;
}

public void setFileId(String fileId) {
this.fileId = fileId;
}

public Option<String> getBasePath() {
Expand Down