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

zhangyue19921010 · 2022-11-10T09:17:14Z

Change Logs

Followed PR #5416

Add a new Executor Type named SIMPLE for hoodie records writer.

This Simple executor is Single Writer and Single Reader mode. Also this SimpleHoodieExecutor has no inner message queue and no inner lock which can consume and writing records from iterator directly.

Advantages: There is no need for additional memory and cpu resources due to lock or multithreading.

Disadvantages: lost some benefits such as speed limit. And can not de-coupe the network read (shuffle read) and network write (writing objects/files to storage) anymore which may cause a little lower throughput compared with DisruptorExecutor.

Also I did a quick benchmark using hoodie benchmark framework org.apache.spark.sql.execution.benchmark.BoundInMemoryExecutorBenchmark

Minimize the impact of producers and consumers efficiency issue as much as possible to focus on testing the throughput limit of the inner queue

The result are

OpenJDK 64-Bit Server VM 1.8.0_342-b07 on Linux 5.10.62-55.141.amzn2.x86_64
Intel(R) Xeon(R) Platinum 8259CL CPU @ 2.50GHz
COW Ingestion:                            Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
BoundInMemory Executor                            34661          35143         292          0.3        3466.1       1.0X
Simple Executor                                   17347          17796         681          0.6        1734.7       2.0X
Disruptor Executor                                15803          16535         936          0.6        1580.3       2.2X

this Simple Executor has good throughput and minimal resource usage.

Also add the corresponding UTs

Of course we need to unify all these executor/messageQueue related UTs as much as possible.
Create a new Tickets https://issues.apache.org/jira/browse/HUDI-5106
Will do it as next step

Impact

No impact

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

low

Documentation Update

no

Contributor's checklist

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

zhangyue19921010 · 2022-11-11T03:31:53Z

Hi @alexeykudinkin and @nsivabalan
Sorry to bother u.
Would u mind to take a further look at this PR?
Really appreciate it if it's possible!

zhangyue19921010 · 2022-11-17T08:54:26Z

Hi @alexeykudinkin Sorry to bother u.
As we discuss in #6843
I review our executor and queue related interface.

Just keep on common method in HoodieMessageQueue which is necessary.

So would u mind to take a look at your convert?

Really appreciate it!

alexeykudinkin

@zhangyue19921010 thanks for following up on this one!

alexeykudinkin · 2022-11-18T06:10:19Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieExecutor.java

+ * Advantages: there is no need for additional memory and cpu resources due to lock or multithreading.
+ * Disadvantages: lost some benefits such as speed limit. And maybe lower throughput.
+ */
+public class SimpleHoodieExecutor<I, O, E> extends HoodieExecutorBase<I, O, E> {


Let's actually simplify this even further and just inherit from HoodieExecutor

alexeykudinkin · 2022-11-18T19:33:52Z

@zhangyue19921010 let's also rebase on top of this one: #7238

alexeykudinkin · 2022-11-18T19:36:49Z

For those queries about relative performance of BIMQ/Disruptor/Simple on EMR cluster, i've captured this in: #6843 (comment)

Pasting the results in here:

#
# BoundedInMemoryQueue (baseline, master)
#

==================================================
Total time taken by all rounds (hudi): 5119338
Per round: List(204855, 504984, 585545, 646676, 516914, 554007, 510029, 533198, 554438, 508692)
==================================================


#
# Disruptor (master)
# Performance Improvement (from baseline): ~2%
#

===================================================
Total time taken by all rounds (hudi): 5027185
Per round: List(142641, 450997, 544476, 576815, 552873, 585657, 526764, 552790, 563460, 530712)
==================================================

#
# Simple (no queue)
# Performance Improvement (from baseline): ~26%
#

==================================================
Total time taken by all rounds (hudi): 3798022
Per round: List(189704, 467253, 512628, 468869, 363557, 361869, 351934, 357554, 364299, 360355)
==================================================

codope · 2022-12-07T13:54:52Z

@zhangyue19921010 Can you please rebase?

zhangyue19921010 · 2022-12-08T09:50:33Z

Hi @alexeykudinkin and @codope Rebased this PR.
Also re-run this benchmark and get the same result.

alexeykudinkin · 2022-12-08T17:30:49Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieMessageQueue.java

+/**
+ * Wrapper of input records iterator
+ */
+public class SimpleHoodieMessageQueue<I, O> implements HoodieMessageQueue<I, O>, Iterable<O> {


@zhangyue19921010 i think we need to clean this one up -- we don't actually need it as we use to only pass an iterator essentially

alexeykudinkin · 2022-12-08T17:33:22Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieExecutor.java

+ * Advantages: there is no need for additional memory and cpu resources due to lock or multithreading.
+ * Disadvantages: lost some benefits such as speed limit. And maybe lower throughput.
+ */
+public class SimpleHoodieExecutor<I, O, E> extends BaseHoodieQueueBasedExecutor<I, O, E> {


This is not queue-based, let's make it inherit from HoodieExecutor instead

Do you mind cleaning up some generics which are not used anymore w/in HoodieExecutor?

All done. Also verify the performance.
Thanks for your review!

zhangyue19921010 · 2022-12-11T09:33:39Z

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

+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestSimpleExecutionInSpark extends HoodieClientTestHarness {


Hi @@alexeykudinkin
Not involved testInterruptExecutor in this SimpleExecutionSpark Test.
Bcz there is no inner Thread pool in this new SimpleExecutor so there is nothing to interrupt right?

Is that looks good to u ?

Yeah, that makes sense

zhangyue19921010 · 2022-12-11T15:32:18Z

@hudi-bot run azure

alexeykudinkin · 2022-12-12T17:50:05Z

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

+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestSimpleExecutionInSpark extends HoodieClientTestHarness {


Yeah, that makes sense

alexeykudinkin · 2022-12-12T17:51:15Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieExecutor.java

+  // records iterator
+  protected final Iterator<I> it;
+  private final Function<I, O> transformFunction;
+  private final Runnable preExecuteRunnable;


We don't need this one (this one is only required for these that spin up new threads)

alexeykudinkin · 2022-12-12T17:53:01Z

...src/test/scala/org/apache/spark/sql/execution/benchmark/BoundInMemoryExecutorBenchmark.scala

-   *
+   * BoundInMemory Executor                            34661          35143         292          0.3        3466.1       1.0X
+   * Simple Executor                                   17347          17796         681          0.6        1734.7       2.0X
+   * Disruptor Executor                                15803          16535         936          0.6        1580.3       2.2X


Let's amend the commentary at the top that this benchmark has been run w/ unconstrained parallelism which obviously is beneficial to Disruptor more than it's for Simple

alexeykudinkin · 2022-12-12T19:47:46Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieExecutor.java

+    return isWriteDone.get();
+  }
+
+  public boolean isRunning() {


I don't think we need this one (we only use it in test, and can actually avoid that assertion altogether, since we're asserting # of records)

alexeykudinkin · 2022-12-12T19:48:42Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieExecutor.java

+  private final Function<I, O> transformFunction;
+  private final Runnable preExecuteRunnable;
+  private final AtomicBoolean isWriteDone = new AtomicBoolean(false);
+  private final AtomicBoolean isShutdown = new AtomicBoolean(false);


There's actually no state for us to clean up so we can just kill these 2

alexeykudinkin · 2022-12-12T19:51:38Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieExecutor.java

+      LOG.info("Starting consumer, consuming records from the records iterator directly");
+      preExecuteRunnable.run();
+      while (it.hasNext()) {
+        if (isShutdown.get()) {


I don't think we need this check since

We don't expect to shutdown executor in the middle of iteration (in the normal course)

In case of exception this would be interrupted naturally

alexeykudinkin · 2022-12-12T19:55:44Z

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

+    try {
+      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getTransformFunction(HoodieTestDataGenerator.AVRO_SCHEMA), getPreExecuteRunnable());
+      int result = exec.execute();
+      // It should buffer and write 100 records


There's no buffering in here, let's update this one

zhangyue19921010 · 2022-12-13T13:34:56Z

Hey @alexeykudinkin All comments addressed. PTAL

alexeykudinkin

Thanks for addressing the feedback @zhangyue19921010!

LGTM, left minor comments regarding the test, PTAL

alexeykudinkin · 2022-12-13T20:45:15Z

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

+    cleanupResources();
+  }
+
+  private Runnable getPreExecuteRunnable() {


I don't think we need this one

alexeykudinkin · 2022-12-13T20:45:57Z

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

+      // collect all records and assert that consumed records are identical to produced ones
+      // assert there's no tampering, and that the ordering is preserved
+      assertEquals(hoodieRecords, consumedRecords);
+      for (int i = 0; i < hoodieRecords.size(); i++) {


We don't need this for-loop, right?

Yeap. Removed. Sorry for missing this one.

alexeykudinkin · 2022-12-13T20:46:34Z

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

+    });
+
+    HoodieWriteConfig hoodieWriteConfig = mock(HoodieWriteConfig.class);
+    when(hoodieWriteConfig.getDisruptorWriteBufferSize()).thenReturn(Option.of(16));


We need to clean up these

zhangyue19921010 · 2022-12-15T09:54:04Z

Hi @alexeykudinkin All comments are addressed. PTAL.
Also really appreciate for your review!

hudi-bot · 2022-12-15T16:25:13Z

CI report:

0cd32cd UNKNOWN
be53e07 UNKNOWN
b0c2730 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

alexeykudinkin · 2022-12-15T17:19:38Z

Awesome job @zhangyue19921010! Thanks for taking this up from idea to completion!

danny0405 · 2022-12-22T08:49:02Z

hudi-common/src/main/java/org/apache/hudi/common/util/queue/SimpleHoodieExecutor.java

+  @Override
+  public E execute() {
+    checkState(this.consumer.isPresent());
+


Since consumer aleays exists, there is no need to keep it in an Option, we can then avoids the invocation Option#get actually.

@danny0405 addressing this in #7476

… inner message queue (apache#7174) Followed PR apache#5416 Add a new Executor Type named SIMPLE for hoodie records writer. This Simple executor is Single Writer and Single Reader mode. Also this SimpleHoodieExecutor has no inner message queue and no inner lock which can consume and writing records from iterator directly. Advantages: There is no need for additional memory and cpu resources due to lock or multithreading. Disadvantages: lost some benefits such as speed limit. And can not de-coupe the network read (shuffle read) and network write (writing objects/files to storage) anymore which may cause a little lower throughput compared with DisruptorExecutor. Also I did a quick benchmark using hoodie benchmark framework org.apache.spark.sql.execution.benchmark.BoundInMemoryExecutorBenchmark Minimize the impact of producers and consumers efficiency issue as much as possible to focus on testing the throughput limit of the inner queue The result are OpenJDK 64-Bit Server VM 1.8.0_342-b07 on Linux 5.10.62-55.141.amzn2.x86_64 Intel(R) Xeon(R) Platinum 8259CL CPU @ 2.50GHz COW Ingestion: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ BoundInMemory Executor 34661 35143 292 0.3 3466.1 1.0X Simple Executor 17347 17796 681 0.6 1734.7 2.0X Disruptor Executor 15803 16535 936 0.6 1580.3 2.2X this Simple Executor has good throughput and minimal resource usage. Also add the corresponding UTs

yuezhang added 12 commits November 9, 2022 12:22

ready to test

5c1c1fd

ready to test

1e99dac

ready to test

37a6e2d

need more test

cc37167

add more java docs; still need to add more UTs

1e966be

performance test

b4f4135

performance test

7a1e171

add more UTs

ec0c804

finished benchmark

0cd32cd

fix checkstyle

7ed45ff

fix checkstyle

178bf76

fix checkstyle

27316e8

xushiyan assigned alexeykudinkin Nov 11, 2022

xushiyan added the priority:critical Production degraded; pipelines stalled label Nov 11, 2022

zhangyue19921010 mentioned this pull request Nov 12, 2022

[WIP][HUDI-5023] Avoiding using BoundedInMemoryExecutor on the hot-path #6843

Closed

4 tasks

refactor interface HoodieMessageQueue

e2e9203

alexeykudinkin reviewed Nov 18, 2022

View reviewed changes

alexeykudinkin changed the title ~~[HUDI-5190] Consuming records from Iterator directly instead of using inner message queue~~ [HUDI-5190][HUDI-5023] Consuming records from Iterator directly instead of using inner message queue Nov 18, 2022

alexeykudinkin changed the title ~~[HUDI-5190][HUDI-5023] Consuming records from Iterator directly instead of using inner message queue~~ [HUDI-5023] Consuming records from Iterator directly instead of using inner message queue Nov 30, 2022

alexeykudinkin mentioned this pull request Dec 2, 2022

[HUDI-5023] Cleaning up QueueBasedExecutor impls #7238

Merged

4 tasks

nsivabalan added the release-0.12.2 Patches targetted for 0.12.2 label Dec 6, 2022

yuezhang added 2 commits December 8, 2022 16:06

rebase & change codes

a5c5fa8

rebase

bca3774

alexeykudinkin reviewed Dec 8, 2022

View reviewed changes

yuezhang added 3 commits December 11, 2022 16:53

address comments

8f2f3a5

address comments

c734cef

address comments

db06b33

zhangyue19921010 commented Dec 11, 2022

View reviewed changes

alexeykudinkin reviewed Dec 12, 2022

View reviewed changes

yuezhang added 2 commits December 13, 2022 17:31

address comments

be53e07

address comments

783a5cd

alexeykudinkin approved these changes Dec 13, 2022

View reviewed changes

yuezhang added 2 commits December 15, 2022 17:43

merge from master

74e0742

address comments

b0c2730

alexeykudinkin merged commit 8d13a7e into apache:master Dec 15, 2022

danny0405 reviewed Dec 22, 2022

View reviewed changes

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

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

Uh oh!

Conversation

zhangyue19921010 commented Nov 10, 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!

zhangyue19921010 commented Nov 11, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

zhangyue19921010 commented Nov 17, 2022

Uh oh!

alexeykudinkin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexeykudinkin commented Nov 18, 2022

Uh oh!

alexeykudinkin commented Nov 18, 2022

Uh oh!

codope commented Dec 7, 2022

Uh oh!

zhangyue19921010 commented Dec 8, 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!

zhangyue19921010 Dec 11, 2022 • 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!

zhangyue19921010 commented Dec 11, 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!

zhangyue19921010 commented Dec 13, 2022

Uh oh!

alexeykudinkin left a comment

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!

zhangyue19921010 commented Dec 15, 2022

zhangyue19921010 commented Nov 10, 2022 •

edited

Loading

zhangyue19921010 commented Nov 11, 2022 •

edited

Loading

zhangyue19921010 Dec 11, 2022 •

edited

Loading