[SUPPORT]

[Shagish]

Hi Team

We are facing an issue in our Prod environment for Hoodie table
Caused by: java.io.FileNotFoundException: No such file or directory 's3://XXXXXXXXX/up_md_info/table/df245ac4-eafb-491b-8f5f-fcbb920b30ee-0_20-1773-8703_20221201102406279.parquet'
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.getFileStatus(S3NativeFileSystem.java:521)
at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.getFileStatus(EmrFileSystem.java:613)

The application was running fine, and it was writing to the hoodie table and all sudden the application failed
when we are trying to bring back the application. it run for 5-10 mints and while writing to the file it throws an error.
Below are the details
What table type cow or mor - MOR
What spark version - 3.2.1
What hudi version - 0.11.0
Where r u running spark jobs - In EMR 6.7.0
What is Hadoop version
What were you trying to
The application is a Spark Hoodie streaming Job. It reads the message from the Kafka topic, process the message and then write to hoodie table. The application Runs for a while and later while writing the data to the Hoodie table, it fails with file not found exception
The file which it complains as not found is very old (12/01/2022) parquet file.
What have you tried
We tried with changing the Hoodie properties and restarting the steps, but it is failed

Below is the log details

022-12-22 22:52:32 INFO YarnClusterScheduler:57 - Killing all running tasks in stage 497: Stage cancelled
2022-12-22 22:52:33 INFO DAGScheduler:57 - ResultStage 497 (start at Application.java:101) failed in 2.542 s due to Job aborted due to stage failure: Task 0 in stage 497.0 failed 4 times, most recent failure: Lost task 0.3 in stage 497.0 (TID 762) (ip-10-220-71-253.emr.awsw.cld.ds.dtvops.net executor 2): org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:329)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleInsertPartition(BaseSparkCommitActionExecutor.java:335)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$mapPartitionsAsRDD$a3ab3c4$1(BaseSparkCommitActionExecutor.java:246)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:915)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:915)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:386)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1498)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1408)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1472)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1295)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:384)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:335)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:133)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1474)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.hudi.exception.HoodieException: java.io.FileNotFoundException: No such file or directory 's3://XXXXXXXXX/up_md_info/table/df245ac4-eafb-491b-8f5f-fcbb920b30ee-0_20-1773-8703_20221201102406279.parquet'
at org.apache.hudi.table.action.commit.HoodieMergeHelper.runMerge(HoodieMergeHelper.java:149)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdateInternal(BaseSparkCommitActionExecutor.java:358)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:349)
at org.apache.hudi.table.action.deltacommit.BaseSparkDeltaCommitActionExecutor.handleUpdate(BaseSparkDeltaCommitActionExecutor.java:80)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:322)
... 29 more
Caused by: java.io.FileNotFoundException: No such file or directory 's3://XXXXXXXXX/up_md_info/table/df245ac4-eafb-491b-8f5f-fcbb920b30ee-0_20-1773-8703_20221201102406279.parquet'
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.getFileStatus(S3NativeFileSystem.java:521)
at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.getFileStatus(EmrFileSystem.java:613)
at org.apache.parquet.hadoop.ParquetReader$Builder.build(ParquetReader.java:337)
at org.apache.hudi.io.storage.HoodieParquetReader.getRecordIterator(HoodieParquetReader.java:70)
at org.apache.hudi.table.action.commit.HoodieMergeHelper.runMerge(HoodieMergeHelper.java:134)
... 33 more
11:13
Driver stacktrace:
2022-12-22 22:52:33 INFO DAGScheduler:57 - Job 333 failed: start at Application.java:101, took 2.693215 s
2022-12-22 22:52:33 ERROR MicroBatchExecution:94 - Query [id = d6b8649b-2b4a-43e3-9140-c997af8e1214, runId = 6c44b42b-3bc3-4e1f-a6a5-0cd542e11cd2] terminated with error
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 497.0 failed 4 times, most recent failure: Lost task 0.3 in stage 497.0 (TID 762) (ip-10-220-71-253.emr.awsw.cld.ds.dtvops.net executor 2): org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:329)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleInsertPartition(BaseSparkCommitActionExecutor.java:335)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$mapPartitionsAsRDD$a3ab3c4$1(BaseSparkCommitActionExecutor.java:246)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:915)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:915)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:386)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1498)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1408)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1472)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1295)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:384)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:335)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:133)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1474)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.hudi.exception.HoodieException: java.io.FileNotFoundException: No such file or directory 's3://XXXXXXXXX/up_md_info/table/df245ac4-eafb-491b-8f5f-fcbb920b30ee-0_20-1773-8703_20221201102406279.parquet'
at org.apache.hudi.table.action.commit.HoodieMergeHelper.runMerge(HoodieMergeHelper.java:149)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdateInternal(BaseSparkCommitActionExecutor.java:358)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:349)
at org.apache.hudi.table.action.deltacommit.BaseSparkDeltaCommitActionExecutor.handleUpdate(BaseSparkDeltaCommitActionExecutor.java:80)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:322)
... 29 more
Caused by: java.io.FileNotFoundException: No such file or directory 's3://XXXXXXXXX/up_md_info/table/df245ac4-eafb-491b-8f5f-fcbb920b30ee-0_20-1773-8703_20221201102406279.parquet'
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.getFileStatus(S3NativeFileSystem.java:521)
at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.getFileStatus(EmrFileSystem.java:613)
at org.apache.parquet.hadoop.ParquetReader$Builder.build(ParquetReader.java:337)
at org.apache.hudi.io.storage.HoodieParquetReader.getRecordIterator(HoodieParquetReader.java:70)
at org.apache.hudi.table.action.commit.HoodieMergeHelper.runMerge(HoodieMergeHelper.java:134)

[yihua]

Hi @Shagish thanks for raising this issue. Could you share the write configs of your Hudi Spark job? One possibility is that the metadata table might be out of sync with the data table for MOR with async table services in 0.11.0 release. To unblock the pipeline, you can try to disable metadata table with hoodie.metadata.enable=false or try the latest 0.12.2 release to see if the problem goes away.

[Shagish]

Hi @yihua,
Thanks for your response and providing us with the solution.
Here are the Hoodie configuration details which we have in our Prod environment.
hoodie.datasource.hive_sync.table=up_customer_info_2
hoodie.table.name=up_customer_info_2
s3.hudi.base.path=s3://XXXXXXXXXXX/up_customer_info_2/table
checkpointLocation=s3://XXXXXXXXXXX/up_customer_info_2/checkpoint
records.error.path=s3://XXXXXXXXXXX/up_customer_info_2/error
hoodie.datasource.write.recordkey.field=accountID
hoodie.datasource.write.precombine.field=op_ts
hoodie.datasource.hive_sync.enable=true
hoodie.datasource.hive_sync.auto_create_database=false
hoodie.datasource.hive_sync.database=integration
hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.NonPartitionedExtractor
hoodie.index.type=SIMPLE
hoodie.datasource.write.operation=UPSERT
hoodie.datasource.write.table.type=COPY_ON_WRITE
hoodie.datasource.write.hive_style_partitioning=true
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.NonpartitionedKeyGenerator
hoodie.upsert.shuffle.parallelism=20
hoodie.insert.shuffle.parallelism=20
hoodie.cleaner.commits.retained=5
hoodie.metrics.reporter.type=GRAPHITE
hoodie.metrics.graphite.host=XXXXXX.DD.net
hoodie.metrics.graphite.port=9109
hoodie.compact.inline=true
hoodie.embed.timeline.server.reuse.enabled=true
hoodie.write.markers.type=timeline_server_based
hoodie.markers.timeline_server_based.batch.num_threads=20
hoodie.markers.timeline_server_based.batch.interval_ms=50
hoodie.metrics.graphite.metric.prefix=spark_streaming_hudi
hoodie.metrics.on=true

Please review and provide us with your comments

Thanks,
Shagish

[yihua]

The issue description mentions that the table type is MOR (What table type cow or mor - MOR), but the write config shows it's using COW (hoodie.datasource.write.table.type=COPY_ON_WRITE). Could you send over the content of the <table_base_path>/.hoodie/hoodie.properties file to confirm the Hudi table properties?

Have you tried setting hoodie.metadata.enable=false and see if that can get rid of the FileNotFoundException exception?

[yihua]

Also, if sync table services are triggered by the streaming job, the concurrency control configs need to be set. If not, this can also lead to inconsistency between the internal metadata table and the Hudi data table.

To repair this, you can always turn off the metadata table by setting hoodie.metadata.enable=false, let the job run for some time to generate some commits, and then turn on metadata table again by removing hoodie.metadata.enable=false.

[Shagish]

Hi @yihua ,
In the Prod, initially we had the table type as MOR. But later we modified it to COW. And the properties which I had shared are the latest from the Prod Env.

We haven't yet tried with hoodie.metadata.enable=false.
Earlier, since the Job was not running even after trying with other properties option and frequent restart, we had to later modify the table name.
We will try with this metadata option.
hoodie.zip

Please find the attached hoodie properties file.
The hoodie file still reflects the table type as MOR while in the properties we had modified it to COW.

Please review this.

Thanks,
Shagish

[Shagish]

hi @yihua ,

We have included hoodie.metadata.enable=false and didn't find any further issue.

Question - what is the implication if we keep hoodie.metadata.enable=false throughout and won't revert it back to true or remove it.

Thanks,
Shagish

[codope]

@Shagish Thanks for confirming the resolution.
To answer your previous question, if you don't revert metadata config back to true, it won't affect the pipeline at all. However, metadata table brings some performance benefits which you may not be able to take full advantage of. In the upcoming release 0.13.0, we have made some foundational fixes to the metadata table which will make its experience even better.