[SUPPORT] Writing Hudi tables with Flink fails with HFile exceptions
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
**Using Flink version : 1.16
Hudi : 0.13.0**
Trying to write a hudi table with metadata sync enabled to HMS, sample DStream API job
```java
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// generate a source of data.
DataStream fares = env.addSource(new TaxiFareGenerator()).map(
event -> GenericRowData.of(
event.getRideId(),
event.getDriverId(),
event.getTaxiId(),
event.getStartTime(),
event.getTip(),
event.getTolls(),
event.getTotalFare()//,
// event.getPaymentType()
));
String targetTable = "TaxiFare";
String outputPath = String.join("/",basePath, "hudi_mor_2023082101");
Map options = new HashMap<>();
options.put(FlinkOptions.PATH.key(), outputPath);
options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
options.put(FlinkOptions.METADATA_ENABLED.key(), "true");
options.put(FlinkOptions.HIVE_SYNC_ENABLED.key(), "true");
options.put(FlinkOptions.HIVE_SYNC_MODE.key(), HiveSyncMode.HMS.name());
options.put(FlinkOptions.HIVE_SYNC_METASTORE_URIS.key(), "thrift://hive-metastore:9083 ");
HoodiePipeline.Builder builder = HoodiePipeline.builder(targetTable)
.column("rideId BIGINT")
.column("driverId BIGINT")
.column("taxiId BIGINT")
.column("startTime BIGINT")
.column("tip FLOAT")
.column("tolls FLOAT")
.column("totalFare FLOAT")
.pk("driverId")
.options(options);
builder.sink(fares, false);
env.execute("Hudi Table with HMS");
```
The job fails with the following exceptions
```
2023-08-21 14:30:42
org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'stream_write: TaxiFare' (operator 07008715da12a894ffb19d48ffca33da).
at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:617)
at org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$start$0(StreamWriteOperatorCoordinator.java:190)
at org.apache.hudi.sink.utils.NonThrownExecutor.handleException(NonThrownExecutor.java:142)
at org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:133)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.hudi.exception.HoodieException: Executor executes action [initialize instant 20230821213020770] error
... 6 more
Caused by: org.apache.hudi.exception.HoodieException: Failed to update metadata
at org.apache.hudi.client.HoodieFlinkTableServiceClient.writeTableMetadata(HoodieFlinkTableServiceClient.java:181)
at org.apache.hudi.client.HoodieFlinkWriteClient.writeTableMetadata(HoodieFlinkWriteClient.java:279)
at org.apache.hudi.client.BaseHoodieWriteClient.commit(BaseHoodieWriteClient.java:282)
at org.apache.hudi.client.BaseHoodieWriteClient.commitStats(BaseHoodieWriteClient.java:233)
at org.apache.hudi.client.HoodieFlinkWriteClient.commit(HoodieFlinkWriteClient.java:111)
at org.apache.hudi.client.HoodieFlinkWriteClient.commit(HoodieFlinkWriteClient.java:74)
at org.apache.hudi.client.BaseHoodieWriteClient.commit(BaseHoodieWriteClient.java:199)
at org.apache.hudi.sink.StreamWriteOperatorCoordinator.doCommit(StreamWriteOperatorCoordinator.java:537)
at org.apache.hudi.sink.StreamWriteOperatorCoordinator.commitInstant(StreamWriteOperatorCoordinator.java:513)
at org.apache.hudi.sink.StreamWriteOperatorCoordinator.commitInstant(StreamWriteOperatorCoordinator.java:484)
at org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$initInstant$6(StreamWriteOperatorCoordinator.java:402)
at org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:130)
... 3 more
Caused by: org.apache.hudi.exception.HoodieUpsertException: Error upsetting bucketType UPDATE for partition :files
at org.apache.hudi.table.action.commit.BaseFlinkCommitActionExecutor.handleUpsertPartition(BaseFlinkCommitActionExecutor.java:203)
at org.apache.hudi.table.action.commit.BaseFlinkCommitActionExecutor.execute(BaseFlinkCommitActionExecutor.java:107)
at org.apache.hudi.table.action.commit.delta.FlinkUpsertPreppedDeltaCommitActionExecutor.execute(FlinkUpsertPreppedDeltaCommitActionExecutor.java:52)
at org.apache.hudi.table.HoodieFlinkMergeOnReadTable.upsertPrepped(HoodieFlinkMergeOnReadTable.java:81)
at org.apache.hudi.client.HoodieFlinkWriteClient.lambda$upsertPreppedRecords$4(HoodieFlinkWriteClient.java:167)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source)
at java.base/java.util.HashMap$ValueSpliterator.forEachRemaining(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.copyInto(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source)
at java.base/java.util.stream.ReduceOps$ReduceTask.doLeaf(Unknown Source)
at java.base/java.util.stream.ReduceOps$ReduceTask.doLeaf(Unknown Source)
at java.base/java.util.stream.AbstractTask.compute(Unknown Source)
at java.base/java.util.concurrent.CountedCompleter.exec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.doInvoke(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.invoke(Unknown Source)
at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateParallel(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source)
at java.base/java.util.stream.ReferencePipeline.collect(Unknown Source)
at org.apache.hudi.client.HoodieFlinkWriteClient.upsertPreppedRecords(HoodieFlinkWriteClient.java:170)
at org.apache.hudi.metadata.FlinkHoodieBackedTableMetadataWriter.commit(FlinkHoodieBackedTableMetadataWriter.java:166)
at org.apache.hudi.metadata.HoodieBackedTableMetadataWriter.processAndCommit(HoodieBackedTableMetadataWriter.java:830)
at org.apache.hudi.metadata.HoodieBackedTableMetadataWriter.update(HoodieBackedTableMetadataWriter.java:897)
at org.apache.hudi.client.HoodieFlinkTableServiceClient.writeTableMetadata(HoodieFlinkTableServiceClient.java:179)
... 14 more
Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.hudi.org.apache.hadoop.hbase.io.hfile.HFile
at org.apache.hudi.common.table.log.block.HoodieHFileDataBlock.serializeRecords(HoodieHFileDataBlock.java:142)
at org.apache.hudi.common.table.log.block.HoodieDataBlock.getContentBytes(HoodieDataBlock.java:111)
at org.apache.hudi.common.table.log.HoodieLogFormatWriter.appendBlocks(HoodieLogFormatWriter.java:158)
at org.apache.hudi.io.HoodieAppendHandle.appendDataAndDeleteBlocks(HoodieAppendHandle.java:464)
at org.apache.hudi.io.HoodieAppendHandle.doAppend(HoodieAppendHandle.java:437)
at org.apache.hudi.table.action.commit.delta.BaseFlinkDeltaCommitActionExecutor.handleUpdate(BaseFlinkDeltaCommitActionExecutor.java:54)
at org.apache.hudi.table.action.commit.BaseFlinkCommitActionExecutor.handleUpsertPartition(BaseFlinkCommitActionExecutor.java:195)
... 37 more
```
Based on the recommendation here, looks like I have to rebuilt hudi flink bundle and then use that, am I correct?
https://hudi.apache.org/docs/faq/#how-can-i-resolve-the-nosuchmethoderror-from-hbase-when-using-hudi-with-metadata-table-on-hdfs
Contributor guide
No contributing guide indexed for this repository
Research direction
Reproduce the sample Flink 1.16 job with Hudi 0.13.0 and metadata sync enabled, then start from HoodieHFileDataBlock.serializeRecords and the HFile initialization failure in the stack trace. Compare the setup with the linked Hudi FAQ; the investigation is complete when the dependency or bundle cause and the required supported configuration are confirmed.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 30/100