[HUDI-5023] Switching default Write Executor type to `SIMPLE` by alexeykudinkin · Pull Request #7476 · apache/hudi

alexeykudinkin · 2022-12-16T00:34:44Z

Change Logs

This change switches default Write Executor to be SIMPLE ie one bypassing reliance on any kind of Queue (either BoundedInMemory or Disruptor's one).

This should considerably trim down on

Runtime (compared to BIMQ)
Compute wasted (compared to BIMQ, Disruptor)

Since it eliminates unnecessary intermediary "staging" of the records in the queue (for ex, in Spark such in-memory enqueueing occurs at the ingress points, ie shuffling), and allows to handle records writing in one pass (even avoiding making copies of the records in the future)

Impact

Users w/ upsert-/insert-heavy payloads should see considerable boost in writing performance.

Risk level (write none, low medium or high below)

Low

Documentation Update

N/A

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

alexeykudinkin · 2022-12-19T20:58:47Z

@hudi-bot run azure

alexeykudinkin · 2022-12-22T23:04:25Z

@hudi-bot run azure

yihua · 2023-01-03T18:55:58Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

    public Builder withWriteBufferSize(int size) {
-      writeConfig.setValue(WRITE_DISRUPTOR_BUFFER_SIZE, String.valueOf(size));
+      writeConfig.setValue(WRITE_EXECUTOR_DISRUPTOR_BUFFER_SIZE, String.valueOf(size));
+      return this;
+    }


nit: If this is not intended to use as a public API, could you remove this and only keep Builder#withWriteExecutorDisruptorWriteBufferSize which has the same functionality?

yihua · 2023-01-03T19:01:28Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

+    }
+
+    @VisibleForTesting
+    public HoodieWriteConfig build(boolean shouldValidate) {


should this method use the default scope/visibility instead of public?

Problem is that it's being used in tests in different packages

OK. We should think about how to hide this, so it is not used by users accidentally.

yihua · 2023-01-03T19:04:48Z

...hudi-spark-client/src/test/java/org/apache/hudi/execution/TestDisruptorExecutionInSpark.java

  private final String instantTime = HoodieActiveTimeline.createNewInstantTime();

+
+  private final HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()


nit: (not required in this PR) the tests using different types of write executors should be generalized in a base class, like TestWriteMarkersBase, instead of duplicating the code. This can be refactored later on.

Agreed. Created a task to follow-up HUDI-5622

yihua · 2023-01-03T19:10:36Z

...nt/hudi-spark-client/src/test/java/org/apache/hudi/execution/TestSimpleExecutionInSpark.java


    try {
-      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getCloningTransformer(HoodieTestDataGenerator.AVRO_SCHEMA));
+      exec = new SimpleExecutor(hoodieRecords.iterator(), consumer, Function.identity());


Still use getTransformer instead of Function.identity() for the tests?

We actually don't need to clone in that case

vinothchandar · 2023-01-09T23:27:00Z

@alexeykudinkin are we claiming that there is no downside to this? esp on large file size? does this still provide same performance. Love to understand benchmarks

hudi-bot · 2023-01-27T08:26:07Z

CI report:

9956563 Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

yihua · 2023-01-27T23:29:04Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

      .withDocumentation("Size of in-memory buffer used for parallelizing network reads and lake storage writes.");

-  public static final ConfigProperty<String> WRITE_DISRUPTOR_BUFFER_SIZE = ConfigProperty
+  public static final ConfigProperty<String> WRITE_EXECUTOR_DISRUPTOR_BUFFER_SIZE = ConfigProperty


(Let's take this in a separate PR) What is the unit of this config (B, KB, or MB)? We should mention the unit in the config naming.

Discussed offline: will follow-up w/ separate PR addressing this

yihua · 2023-01-27T23:30:20Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

+    }
+
+    @VisibleForTesting
+    public HoodieWriteConfig build(boolean shouldValidate) {


OK. We should think about how to hide this, so it is not used by users accidentally.

yihua · 2023-01-27T23:37:12Z

...hudi-spark-client/src/test/java/org/apache/hudi/execution/TestDisruptorExecutionInSpark.java

-      exec = new DisruptorExecutor(hoodieWriteConfig.getDisruptorWriteBufferSize(), hoodieRecords.iterator(), consumer,
-          getCloningTransformer(HoodieTestDataGenerator.AVRO_SCHEMA), Option.of(WaitStrategyFactory.DEFAULT_STRATEGY), getPreExecuteRunnable());
+      exec = new DisruptorExecutor<>(writeConfig.getWriteExecutorDisruptorWriteBufferSize(), hoodieRecords.iterator(), consumer,
+          Function.identity(), WaitStrategyFactory.DEFAULT_STRATEGY, getPreExecuteRunnable());


Should this also use getTransformer(HoodieTestDataGenerator.AVRO_SCHEMA, writeConfig)?

There's no reason to over-complicate this test -- it's goal is to test the Executor not how it's being used

yihua · 2023-01-27T23:37:20Z

...hudi-spark-client/src/test/java/org/apache/hudi/execution/TestDisruptorExecutionInSpark.java

-        getCloningTransformer(HoodieTestDataGenerator.AVRO_SCHEMA), Option.of(WaitStrategyFactory.DEFAULT_STRATEGY), getPreExecuteRunnable());
+    DisruptorExecutor<HoodieRecord, HoodieRecord, Integer>
+        executor = new DisruptorExecutor<>(1024, hoodieRecords.iterator(), consumer,
+        Function.identity(), WaitStrategyFactory.DEFAULT_STRATEGY, getPreExecuteRunnable());


similar here

yihua · 2023-01-27T23:39:03Z

...nt/hudi-spark-client/src/test/java/org/apache/hudi/execution/TestSimpleExecutionInSpark.java


    try {
-      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getCloningTransformer(HoodieTestDataGenerator.AVRO_SCHEMA));
+      exec = new SimpleExecutor<>(hoodieRecords.iterator(), consumer, Function.identity());


Similar here for getTransformer. The production code passes the transfer in by calling getTransformer so let's follow the same in the test code.

yihua · 2023-01-27T23:41:55Z

...nt/hudi-spark-client/src/test/java/org/apache/hudi/execution/TestSimpleExecutionInSpark.java


    try {
-      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getCloningTransformer(HoodieTestDataGenerator.AVRO_SCHEMA));
+      exec = new SimpleExecutor<>(hoodieRecords.iterator(), consumer, Function.identity());


similar here

yihua · 2023-01-27T23:41:58Z

...nt/hudi-spark-client/src/test/java/org/apache/hudi/execution/TestSimpleExecutionInSpark.java

-    SimpleHoodieExecutor<HoodieRecord, Tuple2<HoodieRecord, Option<IndexedRecord>>, Integer> exec =
-        new SimpleHoodieExecutor(iterator, consumer, getCloningTransformer(HoodieTestDataGenerator.AVRO_SCHEMA));
+    SimpleExecutor<HoodieRecord, HoodieRecord, Integer> exec =
+        new SimpleExecutor<>(iterator, consumer, Function.identity());


similar here

yihua · 2023-01-27T23:48:32Z

@alexeykudinkin let's make sure we cover different scenarios (including the different base file sizes) in the benchmarking, and that there is no regression in common use cases before landing this PR.

yihua

LGTM

This change switches default Write Executor to be SIMPLE ie one bypassing reliance on any kind of Queue (either BoundedInMemory or Disruptor's one). This should considerably trim down on Runtime (compared to BIMQ) Compute wasted (compared to BIMQ, Disruptor) Since it eliminates unnecessary intermediary "staging" of the records in the queue (for ex, in Spark such in-memory enqueueing occurs at the ingress points, ie shuffling), and allows to handle records writing in one pass (even avoiding making copies of the records in the future)

…#7476) This change switches default Write Executor to be SIMPLE ie one bypassing reliance on any kind of Queue (either BoundedInMemory or Disruptor's one). This should considerably trim down on Runtime (compared to BIMQ) Compute wasted (compared to BIMQ, Disruptor) Since it eliminates unnecessary intermediary "staging" of the records in the queue (for ex, in Spark such in-memory enqueueing occurs at the ingress points, ie shuffling), and allows to handle records writing in one pass (even avoiding making copies of the records in the future)

alexeykudinkin requested a review from xushiyan December 19, 2022 20:58

alexeykudinkin changed the title ~~[MINOR] Switching default Write Executor type to SIMPLE~~ [HUDI-5023] Switching default Write Executor type to SIMPLE Dec 19, 2022

alexeykudinkin requested a review from yihua December 21, 2022 18:44

alexeykudinkin added priority:blocker Production down; release blocker area:performance Performance optimizations labels Dec 21, 2022

alexeykudinkin assigned yihua Dec 21, 2022

alexeykudinkin mentioned this pull request Dec 22, 2022

[HUDI-5023] Consuming records from Iterator directly instead of using inner message queue #7174

Merged

4 tasks

yihua reviewed Jan 3, 2023

View reviewed changes

Alexey Kudinkin added 11 commits January 25, 2023 20:21

Tidying up config

4611e79

Switching default executor type to SIMPLE

b91d344

Fixing compilation

0c4ab58

Tidying up BIMQ, Disruptor

a3544b9

Cleaned up SimpleExecutor

b6c60a0

Cleaned up unnecessary Option wrapping in DisruptorExecutor

d9c17b3

Avoid unnecesary cloning of the records when using SimpleExecutor

e3ea066

Fixing compilation

bb322f5

Make HoodieWriteConfig validation optional for tests

6a6bced

Fixing tests

0173a0e

Tidying up

dcdbe36

alexeykudinkin force-pushed the ak/wrt-exec-swc branch 2 times, most recently from 4c401fc to 5935539 Compare January 27, 2023 00:28

Fixing tests

9956563

alexeykudinkin force-pushed the ak/wrt-exec-swc branch from 5935539 to 9956563 Compare January 27, 2023 04:56

yihua reviewed Jan 27, 2023

View reviewed changes

yihua approved these changes Jan 28, 2023

View reviewed changes

alexeykudinkin merged commit ff590c6 into apache:master Jan 28, 2023

		private final String instantTime = HoodieActiveTimeline.createNewInstantTime();


		private final HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()

Conversation

alexeykudinkin commented Dec 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Change Logs

Impact

Risk level (write none, low medium or high below)

Documentation Update

Contributor's checklist

Uh oh!

alexeykudinkin commented Dec 19, 2022

Uh oh!

alexeykudinkin commented Dec 22, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vinothchandar commented Jan 9, 2023

Uh oh!

hudi-bot commented Jan 27, 2023

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexeykudinkin Jan 27, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yihua commented Jan 27, 2023

Uh oh!

yihua left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

alexeykudinkin commented Dec 16, 2022 •

edited

Loading

alexeykudinkin Jan 27, 2023 •

edited

Loading