Can you concurrently write data to Apache Hudi w/o any lock provider?

Recently, one of the OSS hudi user reached out asking if they can concurrently write to a single hudi table from multiple writers without needing any locks. And they called out that its an immutable workload. Generally for any multi-writing capability, hudi recommends to enable lock configurations. But this was an interesting ask and we explored and found a way out. So, thought will share it with the broader community.

Need for lock provider with concurrent writers:

Might be obvious for some, but may not be for all. So, let’s first take a look at why we need lock providers when concurrent writers write to Hudi or any table format for that matter. If two concurrent writers modify the same set of data, we can allow only one of them to succeed and abort the other since there are conflicts atleast with Optimistic Concurrency control (OCC). We can try to design and implement MVCC based one, but we are not yet there. So, parking that aside, just with pure OCC, no two concurrent writers writing overlapping data can succeed. So, for conflict resolution and for some of the table management services, we need to take locks since only one of them can operate the critical section at any point in time. So, we employ a lock provider to ensure such conflict resolution and table management services are coordinated among the two writers.

So, let’s unpack the reasoning more

1. For conflict resolution purposes so that we don’t let two writers write overlapping data succeed.
2. For table management services like cleaning, archival, clustering etc, we need coordination among regular writers.

So, what incase the two reasons quoted above can be relaxed?

• If your is an immutable workload or different writers write to completely different partitions, then we really don’t need to do any conflict resolution. Obviously, it’s on the user that they take a call here and claim that none of the writers overlap, since hudi may not do any conflict resolution.
• What incase we disable table services for all writers except one.

In summary, both the reasons called out has become moot now. So, can we still achieve this ? Yes, we can definitely do that.

Immutable workloads:

Immutable workloads is the key here. So, we suggested them to use “bulk_insert” as the operation type since its equivalent to writing to a parquet table. There is no index lookup, no small file management, and hence no two writers can collide in any way.

Table services:

Hudi has a global config to disable table services if need be(“hoodie.table.services.enabled”). By default the config is set to true and hence every writer you spin up might be doing table services. But here, we can use this config to disable for all writers except one.

Metadata table:

We have to disable metadata table since we have a prerequisite that the metadata table needs a lock if there are more than one writer.

Here is a pictorial representation of how we can achieve this

In essence, one of the writer will take up ingestion along w/ all table services, while all other writers will only does ingestion which may not overlap with any other writer.

So, putting it altogether, here is what the configs might look like for two writers.

Writer1:

Ignoring the typical mandatory fields like record key, table name etc. these are the configs that one has to set for writer 1.

option("hoodie.datasource.write.operation","bulk_insert"). 
option("hoodie.write.concurrency.mode","OPTIMISTIC_CONCURRENCY_CONTROL").
option("hoodie.cleaner.policy.failed.writes","LAZY").
option("hoodie.write.lock.provider","org.apache.hudi.client.transaction.lock.InProcessLockProvider").
option("hoodie.metadata.enable","false").

As you could notice, we have enabled InProcessLockProvider and set operation type to “bulk_insert” and have disabled metadata.

So, this writer will take care of table services like cleaning and archival.

Writer2:

Let’s take a look at what configs to set for writer2.

option("hoodie.datasource.write.operation","bulk_insert"). 
option("hoodie.cleaner.policy.failed.writes","LAZY"). 
option("hoodie.metadata.enable","false").
option("hoodie.table.services.enabled","false").

As you could notice, we have disabled table services and metadata table and have set the operation type to “bulk_insert”. Hence, all writer2 does is to ingest new data into the table w/o worrying about any of the table services.

Small file management:

If you wish to leverage small file management, you can set the operation type to “insert” for writer1 as well. Care should be taken if you wish to “insert” as the operation type to all writes. If each of them write to different partitions, it might work. But if they might write to same partitions, it might lead to unexpected consequences and should be avoided.

Alternatively, we can keep the operation type to “bulk_insert”, but enable clustering with writer1 as follows:

option("hoodie.datasource.write.operation","bulk_insert"). 
option("hoodie.write.concurrency.mode","OPTIMISTIC_CONCURRENCY_CONTROL").
option("hoodie.cleaner.policy.failed.writes","LAZY").
option("hoodie.write.lock.provider","org.apache.hudi.client.transaction.lock.InProcessLockProvider").
option("hoodie.metadata.enable","false").
option("hoodie.clustering.inline","true").
option("hoodie.clustering.inline.max.commits","4").

I tried the above set of configs for two concurrent spark datasource writers and tested it w/ 100+ commits w/ aggressive cleaner and archival setting. Also induced failures and things were intact. Data from input matched w/ snapshot read from hudi for both writers.

Conclusion:

If your use-case fits the constraints called out earlier, this will be very powerful in increasing the throughput of your writes to Hudi. Not having to deal with managing your infra for lock providers will ease the operational burden as well. I hope this blog benefits some Hudi users.