apache / apache/hudi

[SUPPORT] Facing org.apache.parquet.io.ParquetDecodingException: Failed to read N bytes on Hudi 0.14.0 with offline clustering

Open
#9,942 5 comments 0 reactions 0 assignees View on GitHub
area:table-service issue:data-corruption priority:critical
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

**Describe the problem you faced**

Running Hudi offline clustering job for MoR tables on Hudi 0.14.0
Facing `org.apache.parquet.io.ParquetDecodingException: Failed to read 808583168 bytes` errors.
I found a file occurs that error, can be read using `spark.read.parquet`.

Here is all configuration for Hudi

```properties
hoodie.metadata.enable=true
hoodie.metadata.index.bloom.filter.enable=true
hoodie.datasource.hive_sync.use_jdbc=false
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator
hoodie.parquet.small.file.limit=0
hoodie.bloom.index.use.metadata=true
hoodie.metadata.index.column.stats.enable=true
hoodie.datasource.write.recordkey.field=RECORD_KEY
hoodie.clustering.plan.strategy.sort.columns=RECORD_KEY
hoodie.datasource.write.drop.partition.columns=false
hoodie.clustering.plan.strategy.small.file.limit=536870912
hoodie.clustering.plan.strategy.partition.selected=PARTITION_1
hoodie.datasource.write.row.writer.enable=true
hoodie.bloom.index.update.partition.path=true
hoodie.datasource.hive_sync.database=HIVE_DATABASE
hoodie.datasource.hive_sync.table=TABLE_NAME
hoodie.parquet.compression.codec=snappy
hoodie.datasource.hive_sync.partition_fields=PARTITION_FIELD
hoodie.datasource.meta.sync.enable=true
hoodie.datasource.hive_sync.metastore.uris=thrift://THRIFT_URL:PORT
hoodie.clustering.max.parallelism=720
hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor
hoodie.deltastreamer.ingestion.targetBasePath=hdfs://path/to/hudi/table
hoodie.datasource.hive_sync.mode=hms
hoodie.index.type=BLOOM
hoodie.datasource.hive_sync.assume_date_partitioning=false
hoodie.table.name=TABLE_NAME
hoodie.streamer.schemaprovider.registry.url=https://schema-registry.com/path/to/schema
hoodie.datasource.write.reconcile.schema=true
hoodie.meta.sync.client.tool.class=org.apache.hudi.hive.HiveSyncTool
hoodie.streamer.schemaprovider.registry.schemaconverter=
hoodie.clustering.plan.strategy.max.num.groups=2048
hoodie.datasource.write.partitionpath.field=PARTITION_FIELD
hoodie.clustering.plan.strategy.target.file.max.bytes=1073741824
hoodie.datasource.write.hive_style_partitioning=true
```

**To Reproduce**

Steps to reproduce the behavior:

1. Run the offline clustering job with the above config
2. Clustering mode is schedule and execute

**Expected behavior**

The offline clustering job works.

**Environment Description**

* Hudi version : 0.14.0

* Spark version : 3.3.1

* Hive version : 2.3.6

* Hadoop version : 3.3.4

* Storage (HDFS/S3/GCS..) : HDFS

* Running on Docker? (yes/no) : Yes. Spark on Docker

**Additional context**

Add any other context about the problem here.

**Stacktrace**

```
23/10/30 14:27:04 WARN TaskSetManager: Lost task 8.0 in stage 331.0 (TID 649) (YARN_CLUSTER executor 51): org.apache.parquet.io.ParquetDecodingException: Failed to read 808583168 bytes
at org.apache.spark.sql.execution.datasources.parquet.VectorizedPlainValuesReader.getBuffer(VectorizedPlainValuesReader.java:109)
at org.apache.spark.sql.execution.datasources.parquet.VectorizedPlainValuesReader.readBinary(VectorizedPlainValuesReader.java:364)
at org.apache.spark.sql.execution.datasources.parquet.ParquetVectorUpdaterFactory$BinaryUpdater.readValues(ParquetVectorUpdaterFactory.java:725)
at org.apache.spark.sql.execution.datasources.parquet.VectorizedRleValuesReader.readBatchInternal(VectorizedRleValuesReader.java:244)
at org.apache.spark.sql.execution.datasources.parquet.VectorizedRleValuesReader.readBatch(VectorizedRleValuesReader.java:176)
at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.readBatch(VectorizedColumnReader.java:252)
at org.apache.spark.sql.execution.datasources.parquet.VectorizedParquetRecordReader.nextBatch(VectorizedParquetRecordReader.java:316)
at org.apache.spark.sql.execution.datasources.parquet.VectorizedParquetRecordReader.nextKeyValue(VectorizedParquetRecordReader.java:212)
at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:39)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at org.apache.spark.util.random.SamplingUtils$.reservoirSampleAndCount(SamplingUtils.scala:57)
at org.apache.spark.RangePartitioner$.$anonfun$sketch$1(Partitioner.scala:306)
at org.apache.spark.RangePartitioner$.$anonfun$sketch$1$adapted(Partitioner.scala:304)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:907)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:907)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:329)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:136)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)
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: java.io.EOFException
at org.apache.parquet.bytes.SingleBufferInputStream.slice(SingleBufferInputStream.java:116)
at org.apache.spark.sql.execution.datasources.parquet.VectorizedPlainValuesReader.getBuffer(VectorizedPlainValuesReader.java:107)
... 30 more
```

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with the listed Hudi 0.14.0 offline clustering configuration and reproduce the schedule-and-execute job on Spark 3.3.1. Compare the failing file with the reported successful spark.read.parquet read, then trace the supplied ParquetDecodingException through Spark's vectorized reader. Done means the offline clustering job completes without the reported EOFException.

Written by the indexing model from the issue text.

Assessment

Tech stack
docker, hadoop, java, spark
Domain
data-engineering, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Needs clarification
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.