You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2025/03/22 05:00:35 UTC
(hudi) branch master updated: [HUDI-9198] Support rate limit for append mode (#12999)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 53216051b1c [HUDI-9198] Support rate limit for append mode (#12999)
53216051b1c is described below
commit 53216051b1c96f4944c7bbaf07c8cbb6c9b65398
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Sat Mar 22 13:00:25 2025 +0800
[HUDI-9198] Support rate limit for append mode (#12999)
---
.../AppendWriteFunctionWithRateLimit.java} | 19 ++++++++---------
...riteOperator.java => AppendWriteFunctions.java} | 24 ++++++++++++----------
.../hudi/sink/append/AppendWriteOperator.java | 2 +-
.../RowDataToHoodieFunctionWithRateLimit.java | 2 +-
4 files changed, 24 insertions(+), 23 deletions(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunctionWithRateLimit.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithRateLimit.java
similarity index 75%
copy from hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunctionWithRateLimit.java
copy to hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithRateLimit.java
index 6ba267e303f..2e764dd883e 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunctionWithRateLimit.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithRateLimit.java
@@ -16,23 +16,22 @@
* limitations under the License.
*/
-package org.apache.hudi.sink.transform;
+package org.apache.hudi.sink.append;
-import org.apache.hudi.client.model.HoodieFlinkInternalRow;
import org.apache.hudi.common.util.RateLimiter;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
import java.util.concurrent.TimeUnit;
/**
- * Function that transforms RowData to a HoodieRecord with RateLimit.
+ * Append write function with configurable rate limit.
*/
-public class RowDataToHoodieFunctionWithRateLimit<I extends RowData, O extends HoodieFlinkInternalRow>
- extends RowDataToHoodieFunction<I, O> {
+public class AppendWriteFunctionWithRateLimit<I>
+ extends AppendWriteFunction<I> {
/**
* Total rate limit per second for this job.
*/
@@ -43,8 +42,8 @@ public class RowDataToHoodieFunctionWithRateLimit<I extends RowData, O extends H
*/
private transient RateLimiter rateLimiter;
- public RowDataToHoodieFunctionWithRateLimit(RowType rowType, Configuration config) {
- super(rowType, config);
+ public AppendWriteFunctionWithRateLimit(RowType rowType, Configuration config) {
+ super(config, rowType);
this.totalLimit = config.getLong(FlinkOptions.WRITE_RATE_LIMIT);
}
@@ -56,8 +55,8 @@ public class RowDataToHoodieFunctionWithRateLimit<I extends RowData, O extends H
}
@Override
- public O map(I i) throws Exception {
+ public void processElement(I value, Context ctx, Collector<Object> out) throws Exception {
rateLimiter.acquire(1);
- return super.map(i);
+ super.processElement(value, ctx, out);
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctions.java
similarity index 61%
copy from hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteOperator.java
copy to hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctions.java
index ad1a00203e8..c99c13fe34d 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteOperator.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctions.java
@@ -18,24 +18,26 @@
package org.apache.hudi.sink.append;
-import org.apache.hudi.sink.common.AbstractWriteOperator;
-import org.apache.hudi.sink.common.WriteOperatorFactory;
+import org.apache.hudi.configuration.FlinkOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.types.logical.RowType;
/**
- * Operator for {@link AppendWriteFunction}.
- *
- * @param <I> The input type
+ * Utilities for {@link AppendWriteFunction} to handle rate limit if it was set.
*/
-public class AppendWriteOperator<I> extends AbstractWriteOperator<I> {
-
- public AppendWriteOperator(Configuration conf, RowType rowType) {
- super(new AppendWriteFunction<>(conf, rowType));
+public abstract class AppendWriteFunctions {
+ private AppendWriteFunctions() {
}
- public static <I> WriteOperatorFactory<I> getFactory(Configuration conf, RowType rowType) {
- return WriteOperatorFactory.instance(conf, new AppendWriteOperator<>(conf, rowType));
+ /**
+ * Creates a {@link AppendWriteFunction} instance based on the given configuration.
+ */
+ public static <I> AppendWriteFunction<I> create(Configuration conf, RowType rowType) {
+ if (conf.getLong(FlinkOptions.WRITE_RATE_LIMIT) > 0) {
+ return new AppendWriteFunctionWithRateLimit<>(rowType, conf);
+ } else {
+ return new AppendWriteFunction<>(conf, rowType);
+ }
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteOperator.java
index ad1a00203e8..0f8c70ec6bd 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteOperator.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteOperator.java
@@ -32,7 +32,7 @@ import org.apache.flink.table.types.logical.RowType;
public class AppendWriteOperator<I> extends AbstractWriteOperator<I> {
public AppendWriteOperator(Configuration conf, RowType rowType) {
- super(new AppendWriteFunction<>(conf, rowType));
+ super(AppendWriteFunctions.create(conf, rowType));
}
public static <I> WriteOperatorFactory<I> getFactory(Configuration conf, RowType rowType) {
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunctionWithRateLimit.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunctionWithRateLimit.java
index 6ba267e303f..665004d783d 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunctionWithRateLimit.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunctionWithRateLimit.java
@@ -29,7 +29,7 @@ import org.apache.flink.table.types.logical.RowType;
import java.util.concurrent.TimeUnit;
/**
- * Function that transforms RowData to a HoodieRecord with RateLimit.
+ * Function that transforms RowData to a {@code HoodieFlinkInternalRow} with RateLimit.
*/
public class RowDataToHoodieFunctionWithRateLimit<I extends RowData, O extends HoodieFlinkInternalRow>
extends RowDataToHoodieFunction<I, O> {