You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/05/14 22:08:48 UTC
(hudi) 12/28: [HUDI-7529] Resolve hotspots in stream read (#10911)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d8cb589eba17a87b90a4c8e33b6156a742eb2ad7
Author: zhuanshenbsj1 <34...@users.noreply.github.com>
AuthorDate: Sat Mar 23 08:30:07 2024 +0800
[HUDI-7529] Resolve hotspots in stream read (#10911)
---
.../apache/hudi/configuration/OptionsResolver.java | 18 +++++++++
.../partitioner/StreamReadAppendPartitioner.java | 34 ++++++++++++++++
.../StreamReadBucketIndexPartitioner.java | 37 ++++++++++++++++++
.../selector/StreamReadAppendKeySelector.java | 31 +++++++++++++++
.../selector/StreamReadBucketIndexKeySelector.java | 31 +++++++++++++++
.../org/apache/hudi/table/HoodieTableSource.java | 45 ++++++++++++++--------
6 files changed, 179 insertions(+), 17 deletions(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
index c7e77767418..f74f4130dbb 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
@@ -77,6 +77,14 @@ public class OptionsResolver {
return operationType == WriteOperationType.INSERT;
}
+ /**
+ * Returns whether the table operation is 'upsert'.
+ */
+ public static boolean isUpsertOperation(Configuration conf) {
+ WriteOperationType operationType = WriteOperationType.fromValue(conf.getString(FlinkOptions.OPERATION));
+ return operationType == WriteOperationType.UPSERT;
+ }
+
/**
* Returns whether the table operation is 'bulk_insert'.
*/
@@ -142,10 +150,20 @@ public class OptionsResolver {
return FilePathUtils.extractPartitionKeys(conf).length > 0;
}
+ /**
+ * Returns whether the table index is bucket index.
+ */
public static boolean isBucketIndexType(Configuration conf) {
return conf.getString(FlinkOptions.INDEX_TYPE).equalsIgnoreCase(HoodieIndex.IndexType.BUCKET.name());
}
+ /**
+ * Returns whether it is a MERGE_ON_READ table, and updates by bucket index.
+ */
+ public static boolean isMorWithBucketIndexUpsert(Configuration conf) {
+ return isMorTable(conf) && isUpsertOperation(conf) && isBucketIndexType(conf);
+ }
+
public static HoodieIndex.BucketIndexEngineType getBucketEngineType(Configuration conf) {
String bucketEngineType = conf.get(FlinkOptions.BUCKET_INDEX_ENGINE_TYPE);
return HoodieIndex.BucketIndexEngineType.valueOf(bucketEngineType);
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/partitioner/StreamReadAppendPartitioner.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/partitioner/StreamReadAppendPartitioner.java
new file mode 100644
index 00000000000..0d7e94da06f
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/partitioner/StreamReadAppendPartitioner.java
@@ -0,0 +1,34 @@
+package org.apache.hudi.source.filedistribution.partitioner;
+/*
+ * 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.
+ */
+
+import org.apache.flink.api.common.functions.Partitioner;
+
+public class StreamReadAppendPartitioner implements Partitioner<Integer> {
+
+ private final int parallNum;
+
+ public StreamReadAppendPartitioner(int parallNum) {
+ this.parallNum = parallNum;
+ }
+
+ @Override
+ public int partition(Integer splitNum, int maxParallelism) {
+ return splitNum % parallNum;
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/partitioner/StreamReadBucketIndexPartitioner.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/partitioner/StreamReadBucketIndexPartitioner.java
new file mode 100644
index 00000000000..4b5531b67ba
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/partitioner/StreamReadBucketIndexPartitioner.java
@@ -0,0 +1,37 @@
+/*
+ * 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.source.filedistribution.partitioner;
+
+import org.apache.hudi.index.bucket.BucketIdentifier;
+
+import org.apache.flink.api.common.functions.Partitioner;
+
+public class StreamReadBucketIndexPartitioner implements Partitioner<String> {
+
+ private final int parallNum;
+
+ public StreamReadBucketIndexPartitioner(int parallNum) {
+ this.parallNum = parallNum;
+ }
+
+ @Override
+ public int partition(String fileName, int maxParallelism) {
+ return BucketIdentifier.bucketIdFromFileId(fileName) % parallNum;
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/selector/StreamReadAppendKeySelector.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/selector/StreamReadAppendKeySelector.java
new file mode 100644
index 00000000000..de4a5f85f9c
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/selector/StreamReadAppendKeySelector.java
@@ -0,0 +1,31 @@
+/*
+ * 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.source.filedistribution.selector;
+
+import org.apache.hudi.table.format.mor.MergeOnReadInputSplit;
+
+import org.apache.flink.api.java.functions.KeySelector;
+
+public class StreamReadAppendKeySelector implements KeySelector<MergeOnReadInputSplit, Integer> {
+
+ @Override
+ public Integer getKey(MergeOnReadInputSplit mergeOnReadInputSplit) throws Exception {
+ return mergeOnReadInputSplit.getSplitNumber();
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/selector/StreamReadBucketIndexKeySelector.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/selector/StreamReadBucketIndexKeySelector.java
new file mode 100644
index 00000000000..d1db6559659
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/filedistribution/selector/StreamReadBucketIndexKeySelector.java
@@ -0,0 +1,31 @@
+/*
+ * 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.source.filedistribution.selector;
+
+import org.apache.hudi.table.format.mor.MergeOnReadInputSplit;
+
+import org.apache.flink.api.java.functions.KeySelector;
+
+public class StreamReadBucketIndexKeySelector implements KeySelector<MergeOnReadInputSplit, String> {
+
+ @Override
+ public String getKey(MergeOnReadInputSplit mergeOnReadInputSplit) throws Exception {
+ return mergeOnReadInputSplit.getFileId();
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
index b5fdea7a229..02de8b71d12 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
@@ -46,6 +46,10 @@ import org.apache.hudi.source.FileIndex;
import org.apache.hudi.source.IncrementalInputSplits;
import org.apache.hudi.source.StreamReadMonitoringFunction;
import org.apache.hudi.source.StreamReadOperator;
+import org.apache.hudi.source.filedistribution.partitioner.StreamReadAppendPartitioner;
+import org.apache.hudi.source.filedistribution.partitioner.StreamReadBucketIndexPartitioner;
+import org.apache.hudi.source.filedistribution.selector.StreamReadAppendKeySelector;
+import org.apache.hudi.source.filedistribution.selector.StreamReadBucketIndexKeySelector;
import org.apache.hudi.source.prune.DataPruner;
import org.apache.hudi.source.prune.PartitionPruners;
import org.apache.hudi.source.prune.PrimaryKeyPruners;
@@ -207,24 +211,17 @@ public class HoodieTableSource implements
conf, FilePathUtils.toFlinkPath(path), tableRowType, maxCompactionMemoryInBytes, partitionPruner);
InputFormat<RowData, ?> inputFormat = getInputFormat(true);
OneInputStreamOperatorFactory<MergeOnReadInputSplit, RowData> factory = StreamReadOperator.factory((MergeOnReadInputFormat) inputFormat);
- DataStream<MergeOnReadInputSplit> monitorOperatorStream = execEnv.addSource(monitoringFunction, getSourceOperatorName("split_monitor"))
+ SingleOutputStreamOperator<MergeOnReadInputSplit> monitorOperatorStream = execEnv.addSource(monitoringFunction, getSourceOperatorName("split_monitor"))
.uid(Pipelines.opUID("split_monitor", conf))
- .setParallelism(1)
- .setMaxParallelism(1);
- SingleOutputStreamOperator<RowData> source;
- if (OptionsResolver.isAppendMode(HoodieTableSource.this.conf)) {
- source = monitorOperatorStream
- .transform("split_reader", typeInfo, factory)
- .uid(Pipelines.opUID("split_reader", conf))
- .setParallelism(conf.getInteger(FlinkOptions.READ_TASKS));
- } else {
- source = monitorOperatorStream
- .keyBy(MergeOnReadInputSplit::getFileId)
- .transform("split_reader", typeInfo, factory)
- .uid(Pipelines.opUID("split_reader", conf))
- .setParallelism(conf.getInteger(FlinkOptions.READ_TASKS));
- }
- return new DataStreamSource<>(source);
+ .setParallelism(1);
+
+ DataStream<MergeOnReadInputSplit> sourceWithKey = addFileDistributionStrategy(monitorOperatorStream);
+
+ SingleOutputStreamOperator<RowData> streamReadSource = sourceWithKey
+ .transform("split_reader", typeInfo, factory)
+ .uid(Pipelines.opUID("split_reader", conf))
+ .setParallelism(conf.getInteger(FlinkOptions.READ_TASKS));
+ return new DataStreamSource<>(streamReadSource);
} else {
InputFormatSourceFunction<RowData> func = new InputFormatSourceFunction<>(getInputFormat(), typeInfo);
DataStreamSource<RowData> source = execEnv.addSource(func, asSummaryString(), typeInfo);
@@ -234,6 +231,20 @@ public class HoodieTableSource implements
};
}
+ /**
+ * Specify the file distribution strategy based on different upstream writing mechanisms,
+ * to prevent hot spot issues during stream reading.
+ */
+ private DataStream<MergeOnReadInputSplit> addFileDistributionStrategy(SingleOutputStreamOperator<MergeOnReadInputSplit> source) {
+ if (OptionsResolver.isMorWithBucketIndexUpsert(conf)) {
+ return source.partitionCustom(new StreamReadBucketIndexPartitioner(conf.getInteger(FlinkOptions.READ_TASKS)), new StreamReadBucketIndexKeySelector());
+ } else if (OptionsResolver.isAppendMode(conf)) {
+ return source.partitionCustom(new StreamReadAppendPartitioner(conf.getInteger(FlinkOptions.READ_TASKS)), new StreamReadAppendKeySelector());
+ } else {
+ return source.keyBy(MergeOnReadInputSplit::getFileId);
+ }
+ }
+
@Override
public ChangelogMode getChangelogMode() {
// when read as streaming and changelog mode is enabled, emit as FULL mode;