[HUDI-3963] Use Lock-Free Message Queue Disruptor Improving Hoodie Writing Efficiency

[zhangyue19921010]

https://issues.apache.org/jira/browse/HUDI-3963
RFC design : #5567

Change Logs

Abstract

Add Lock-Free executor to improve hoodie writing throughput and optimize execution efficiency.
Disruptor linked: https://lmax-exchange.github.io/disruptor/user-guide/index.html#_introduction

Background

Abstractly, hoodie consumes upstream data (Kafka or S3 files) into the lake is a standard production-consumption model. Currently, Hudi uses LinkedBlockingQueue as a message queue between Producer and Consumer.

However, this lock model may become the bottleneck of application throughput when data volume is much larger. What's worse is that even if we increase the number of the executors, it is still difficult to improve the throughput.

In other words, users may encounter throughput bottlenecks when writing data into hudi in some scenarios, and increasing the physical hardware scale and parameter tuning cannot solve this problem.

Implementation

1. Abstraction: first abstract the produce-consume process into three parts

• Producer interface: Fill the hoodie data into the inner message queue.
• Executor interface: Initialize, schedule and manage producers and consumers, and hold inner message queues. Executor controls the entire life cycle of data production and consumption.
• Consumer interface: Consumes data from inner message queue and write to hudi file.

2. Following the abstract logic in party1, use the lock-free circular queue Disruptor as the inner message queue of the Executor

• Implement the producer abstraction in party1 and customize the DisruptorBasedProducer which will consume and set data to RingBuffer.
• Implement the abstraction of Executor. Also control the life cycle through Ringbuffer information and related APIs. Customize HoodieDaemonThreadFactory to control Thread behavior.
• Implement Consumer abstraction, define DisruptorMessageHandler to control consumption and clear data in Disruptor.

3. New parameters

• hoodie.write.executor.type: Set executor which orchestrates concurrent producers and consumers communicating through a message queue.
  Default value is BOUNDED_IN_MEMORY_EXECUTOR which use a bounded in-memory queue using LinkedBlockingQueue.
  Also users could use DISRUPTOR_EXECUTOR, which use disruptor as a lock free message queue to gain better writing performance.
  Although DISRUPTOR_EXECUTOR is still an experimental feature.
• hoodie.write.buffer.size: The size of the Disruptor Executor ring buffer, must be power of 2
• hoodie.write.wait.strategy: Strategy employed for making DisruptorExecutor wait on a cursor.

4. limitation
   For now, this disruptor executor is only supported for spark insert and spark bulk insert operation. Other operations like spark upsert is still on going.

Also use this new disruptor based executor will consume more cpu and memory resources.

Rollout/Adoption Plan

Default executor is BOUNDED_IN_MEMORY_EXECUTOR which use a bounded in-memory queue using LinkedBlockingQueue same as now.
So there is no impact on existing users.

Test Plan

1. Add UT TestDisruptorMessageQueue and TestDisruptorExecutionInSpark to guard above logic, also validate data correctness.
2. Add Benchmark BoundInMemoryExecutorBenchmark benchmark with BoundInMemoryExecutor(old) and DisruptorExecutior(new option)

This Patch is tested on dev env, I think it's ready for review.
Although need more java docs here.

Here is the simple schema and spark bulk_insert benchmark locally result between old BoundInMemoryExecutor and new DisruptorExecutor getting 2.0X performance improved :)

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                             5557           5607          70          0.2        5556.9       1.0X
Disruptor Executor                                 2758           2778          28          0.4        2757.7       2.0X

I also do several performance tests recently including

Simple schema (several columns) + small amount of data and complex schema (dozens of columns) and large amount of data using bulk_insert.

Simple schema + small amount of data like the bench mark showed, we may get better performance result from 60% to 2X

As for complex schema (dozens of columns) and large amount of data:
1000 partitions, 100,000,000 records, dozens of columns and 20 * r5.12xlarge instants. We can get about 20% performance improved.

Another conclusion is that the more data-producing faster than data consuming, the more benefit we can get from this PR. In this scenario, locking is the performance bottleneck.

For normal scenes, we can get about 20% performance improvement.

Bench Marks in details

data size	column size	performance tuning
1,000,000 records	7	2X(In Local benchmark)
100,000,000 records	15	23% (In real world)

Please read hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/execution/benchmark/BoundInMemoryExecutorBenchmark.scala for more details.

Impact

Writer-core module.

Risk level medium

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

UPDATE:

From our experience, we have two kinds of spark-streaming ingestion job.

1. For aggregated table which has no unique hoodie keys, so that we use insert or bulkinsert+clustering to do this hudi ingestion.
2. For raw data, use upsert operation.

More details for our performance test :
Hudi version: 0.10.1
Spark Version: 3.0.2 spark streaming
Records number per batch(max): 754932000
Schema: 18 columns

{
  "type": "record",
  "name": "f_order_sa_xxxx_hourly_hudi",
  "namespace": "tv.xxx.schemas",
  "fields": [
    {"name": "timestamp", "type": "long", "doc": "category=timestamp"},
    {"name": "network_id", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "xx_id", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "xx_id", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_id", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_id", "type": ["null", "int"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_id", "type": ["null", "int"], "default": null, "doc": "category=dimension"},
    {"name": "xxxx_visibility", "type": ["null", "string"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_owner_id", "type": ["null", "int"], "default": null, "doc": "category=dimension"},
    {"name": "sxxx_endpoint_id", "type": ["null", "int"], "default": null, "doc": "category=dimension"},
    {"name": "xxxx_order_id", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_source", "type": ["null", "int"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_type", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "content_xxx_id", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_publisher_id", "type": ["null", "string"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_order_id", "type": ["null", "long"], "default": null, "doc": "category=dimension"},
    {"name": "xxx_ad_views", "type": ["null", "long"], "default": null, "doc": "category=metric"},
    {"name": "revenue", "type": ["null", "double"], "default": null, "doc": "category=metric"}
  ]
}

Insert/bulk_insert performance Benchmark between BIMQ (baseline) and Disruptor with same kafka input, resources and configs.

BIMQ: used 7.9 min to finish writing parquets.

Disruptor used 5.6 min to finish writing parquets

In terms of Case 1 write performance, Disruptor improved about 29% from 7.9min to 5.6min

Update2

Recently we adopt this disruptor executor into PRD Spark Streaming update pipeline. In CASE2 mentioned before.

As for writing stage performance, we get about 25% performance tuning. From 15s reduced to 11s

More details:
Spark Streaming batch interval: 30s
Records per batch(Avg): 1,361,453
Hudi version: 0.10.1
Spark Version: 3.0.2

BIMQ: used 15s to finish writing parquets.

Disruptor used 11s to finish writing parquets

[zhangyue19921010]

Hi @yihua @nsivabalan and @xushiyan Sorry to bother you. Do you have time to help review this PR?
Really really appreciate it. Thanks a lot :)

[zhangyue19921010]

@hudi-bot run azure

[alexeykudinkin]

Great effort @zhangyue19921010!

We've really come a long way! Thanks for taking up the clean up the Executor hierarchy, really appreciate it!

[alexeykudinkin]

Let's use it here as well

[alexeykudinkin]

Let's make sure we're not dropping generics params

[zhangyue19921010]

All changed.

[alexeykudinkin]

We can actually just do assertEquals(oneList, anotherList)

[zhangyue19921010]

changed!

[alexeykudinkin]

Let's rebase all of our usage of Future onto CompletableFuture (which are much more capable and versatile than the former)

For that you just need to change how you trigger the execution doing:

CompletableFuture.runAsync(() -> ..., executorService);

Also, seems like this method is missing @Override

[zhangyue19921010]

All changed.

[alexeykudinkin]

nit: "executor-queue-producer", "executor-queue-consumer"

[zhangyue19921010]

Changed.

[alexeykudinkin]

I'd suggest to reserve this interface for APIs only exposed to the actual user: in that case such only APIs are execute and close. All the other APIs are implementation detils (setup, startConumers, postAction, etc) and shouldn't be exposed outside actual implementations:

• Let's move all of these to a shared base class
• Let's keep them "protected"

Let's also make this extend AutoCloseable

[zhangyue19921010]

Nice catch!
Just keep

E execute();
boolean isRemaining();
void shutdownNow();
boolean awaitTermination();

in public interface HoodieExecutor<I, O, E> extends Closeable which are used to control the lifecycle for hoodieExecutor for example SparkLazyInsertIterable#computeNext

And move all other function into HoodieExecutorBase and changed to protect except startProducers and getQueue which need to be call outside.

[alexeykudinkin]

@zhangyue19921010 forgot to remove this class?

[alexeykudinkin]

We should throw UnsupportedOperationException in here

[zhangyue19921010]

All changed here.

[alexeykudinkin]

Should this be package-private?

[zhangyue19921010]

It need to be used in TestDisruptorMessageQueue#testCompositeProducerRecordReading :<

[nsivabalan]

I don't think its a good practice to make methods public just for the purpose of testing. Can we please avoid that.

[zhangyue19921010]

Sure. Use java reflection.

[alexeykudinkin]

We can simplify this to

EventTranslator<HoodieDisruptorEvent> translator = (event, sequence) -> event.set(applied);

[zhangyue19921010]

Sure thing Changed!

[zhangyue19921010]

@hudi-bot run azure

[alexeykudinkin]

@zhangyue19921010 please rebase on the latest master. Test failures you observe have already been addressed on master.

[zhangyue19921010]

@zhangyue19921010 please rebase on the latest master. Test failures you observe have already been addressed on master.

Hi alexeykudinkin Thanks a lot for reminding. rebase to master and all checks are green
Also comments are addressed. PTAL ：）

[alexeykudinkin]

@zhangyue19921010 thank you very much for following up on all of the feedback!

We're waiting on @nsivabalan to give us a green-light from his end, and then we should be good to go

[zhangyue19921010]

@zhangyue19921010 thank you very much for following up on all of the feedback!

We're waiting on @nsivabalan to give us a green-light from his end, and then we should be good to go

Really appreciate it for your efforts here! @alexeykudinkin and @nsivabalan

[nsivabalan]

will review by tomorrow.

[nsivabalan]

mostly minor comments. once addressed, we are good to land.

[nsivabalan]

can we add other possible values in the documentation.

[zhangyue19921010]

Sure thing. Added.

[nsivabalan]

this is confusing w/ already existing one (WRITE_BUFFER_LIMIT_BYTES_VALUE). Can we add units here. initially I thought this is also in bytes. but looks like its ring buffer size. We can also add "ring buffer" or "disruptor" to the name since this is applicable only to disruptor based queue.

[zhangyue19921010]

Sure, units to bytes and changed name to getDisruptorWriteBufferSize()

[nsivabalan]

not required in this patch. But do you think we can have one set of tests and parametrize two diff queues. looking to reuse code. Bcoz, the way we test both bounded in memory queue and disruptor based one are same. just the initialization differs. but essentially its a functional testing of the queue.
let me know wdyt.

[zhangyue19921010]

Nice Idea actually. We should keep the code as concise as possible.
https://issues.apache.org/jira/browse/HUDI-5106
Will working on it after this patch landed :)

[nsivabalan]

can we try to use protected wherever applicable. do we really need public for all these instance variables ?

[zhangyue19921010]

Sure thing. Just go through all the PR and changed wherever applicable.

[nsivabalan]

previously preExecuteRunnable was executed both in producer as well as consumers. W/ this patch, I see we are executing it only for consumers? is that intentional ?
also, I only see it getting executed in BoundedInMemory and not in Distruptor based impl. can you check on that please.

[zhangyue19921010]

Actually, we need to do preExecuteRunnable in Producer/Consumer thread of BoundedInMemory/DisruptorQueue

The original implementation is indeed a bit confusing, and it is simplified here.

expand the CustomizedThreadFactory and will do preExecuteRunnable in each thread if it wasn't null.

Also use this CustomizedThreadFactory to build producerExecutorService and consumerExecutorService

So that we don't need to call preExecuteRunnable manually.

[nsivabalan]

gotcha.

[nsivabalan]

incase of shutdownNow, we shutdown the executorservices and then do queue. close.
where as here, we close queue first followed by shutting down executor services.
may I know whats the right approach. can we fix align.

[zhangyue19921010]

Nice catch. changed.
We align to shutdown the executor services and then do queue.close()

[nsivabalan]

I don't think its a good practice to make methods public just for the purpose of testing. Can we please avoid that.

[nsivabalan]

isn't this ExecutorType.values() ?

[zhangyue19921010]

Sure, changed!

[nsivabalan]

again, lets make protected or package private whereever applicbale.
and lets not make a method public just for testing purpose.

[zhangyue19921010]

All changed. Just go through all the codes in this PR.
If there were anything missed, please let me know. Thanks！

[nsivabalan]

minor. does this refer to total size, or pending size. not apparent from the name.

[zhangyue19921010]

Changed the java docs. it refers to Returns the number of elements in this queue.
Really appreciate for your reviewing here :)

[hudi-bot]

CI report:

• b838e1f UNKNOWN
• 6114ee2 UNKNOWN
• 92760db UNKNOWN
• 0fc24d3 UNKNOWN
• 298f66d UNKNOWN
• 988e14a Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[zhangyue19921010]

hi @nsivabalan Thanks a lot for your review. All comments are addressed. please take a look ：）

[zhangyue19921010]

Also all the Azure jobs are passed now :)

[nsivabalan]

gotcha.

[nsivabalan]

Great job buddy for patiently addressing all comments !

[zhangyue19921010]

Great job buddy for patiently addressing all comments !

Thanks a lot for your help! @alexeykudinkin and @nsivabalan