You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "xicm (via GitHub)" <gi...@apache.org> on 2024/03/19 03:11:31 UTC
[I] [SUPPORT] Archived parquet file lenght is 0 when spark do streaming read [hudi]
xicm opened a new issue, #10881:
URL: https://github.com/apache/hudi/issues/10881
**_Tips before filing an issue_**
- Have you gone through our [FAQs](https://hudi.apache.org/learn/faq/)?
- Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.
- If you have triaged this as a bug, then file an [issue](https://issues.apache.org/jira/projects/HUDI/issues) directly.
**Describe the problem you faced**
```
24/03/19 10:58:34 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 1000 milliseconds, but spent 1141 milliseconds
24/03/19 10:58:38 ERROR MicroBatchExecution: Query [id = 3b464aa8-542d-4eea-b45a-4db8f2a8c8df, runId = d257c670-6a0c-49a8-af65-8cd40f32f1fc] terminated with error
org.apache.hudi.exception.HoodieException: org.apache.hudi.exception.HoodieException: unable to read next record from parquet file
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:593)
at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677)
at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735)
at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160)
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:174)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
at org.apache.hudi.common.table.timeline.HoodieArchivedTimeline.loadInstants(HoodieArchivedTimeline.java:267)
at org.apache.hudi.common.table.timeline.CompletionTimeQueryView.load(CompletionTimeQueryView.java:323)
at org.apache.hudi.common.table.timeline.CompletionTimeQueryView.<init>(CompletionTimeQueryView.java:99)
at org.apache.hudi.common.table.timeline.CompletionTimeQueryView.<init>(CompletionTimeQueryView.java:84)
at org.apache.hudi.common.table.view.AbstractTableFileSystemView.init(AbstractTableFileSystemView.java:121)
at org.apache.hudi.common.table.view.HoodieTableFileSystemView.init(HoodieTableFileSystemView.java:115)
at org.apache.hudi.common.table.view.HoodieTableFileSystemView.<init>(HoodieTableFileSystemView.java:109)
at org.apache.hudi.common.table.view.HoodieTableFileSystemView.<init>(HoodieTableFileSystemView.java:100)
at org.apache.hudi.common.table.view.HoodieTableFileSystemView.<init>(HoodieTableFileSystemView.java:179)
at org.apache.hudi.MergeOnReadIncrementalRelation.collectFileSplits(MergeOnReadIncrementalRelation.scala:110)
at org.apache.hudi.MergeOnReadIncrementalRelation.collectFileSplits(MergeOnReadIncrementalRelation.scala:46)
at org.apache.hudi.HoodieBaseRelation.buildScan(HoodieBaseRelation.scala:371)
at org.apache.spark.sql.hudi.streaming.HoodieStreamSource.getBatch(HoodieStreamSource.scala:177)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$3(MicroBatchExecution.scala:549)
at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:293)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at org.apache.spark.sql.execution.streaming.StreamProgress.foreach(StreamProgress.scala:27)
at scala.collection.TraversableLike.flatMap(TraversableLike.scala:293)
at scala.collection.TraversableLike.flatMap$(TraversableLike.scala:290)
at org.apache.spark.sql.execution.streaming.StreamProgress.flatMap(StreamProgress.scala:27)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$2(MicroBatchExecution.scala:545)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:375)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:373)
at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:68)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runBatch(MicroBatchExecution.scala:545)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:256)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:375)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:373)
at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:68)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:219)
at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:213)
at org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:307)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:285)
at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:208)
Caused by: org.apache.hudi.exception.HoodieException: unable to read next record from parquet file
at org.apache.hudi.common.util.ParquetReaderIterator.hasNext(ParquetReaderIterator.java:54)
at org.apache.hudi.common.table.timeline.HoodieArchivedTimeline.lambda$loadInstants$8(HoodieArchivedTimeline.java:272)
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1067)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1703)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:172)
Caused by: java.lang.RuntimeException: /tmp/hoodie/hudi40/.hoodie/archived/20240319105516693_20240319105559553_0.parquet is not a Parquet file (length is too low: 0)
at org.apache.parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:539)
at org.apache.parquet.hadoop.ParquetFileReader.<init>(ParquetFileReader.java:776)
at org.apache.parquet.hadoop.ParquetFileReader.open(ParquetFileReader.java:657)
at org.apache.parquet.hadoop.ParquetReader.initReader(ParquetReader.java:152)
at org.apache.parquet.hadoop.ParquetReader.read(ParquetReader.java:135)
at org.apache.hudi.common.util.ParquetReaderIterator.hasNext(ParquetReaderIterator.java:49)
... 11 more
```
**To Reproduce**
Steps to reproduce the behavior:
1. start a write task
```
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.apache.spark.sql.functions.{col, lit, rand}
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
val tableName = "hudi40"
val basePath = s"/tmp/hoodie/$tableName"
val ckp = "/tmp/ckp/$tableName"
val hudiOptions = Map(
"hoodie.table.name" -> tableName,
"hoodie.datasource.write.table.type" -> "MERGE_ON_READ",
"hoodie.datasource.write.operation" -> "upsert",
"hoodie.datasource.write.recordkey.field" -> "id",
"hoodie.datasource.write.precombine.field" -> "ts",
"hoodie.datasource.write.partitionpath.field" -> "part",
"hoodie.datasource.hive_sync.enable" -> "false",
"hoodie.datasource.meta.sync.enable" -> "false",
"hoodie.embed.timeline.server" -> "false",
"hoodie.index.type" -> "BUCKET",
"hoodie.bucket.index.num.buckets" -> "2",
"path" -> basePath,
"checkpointLocation" -> ckp
)
val df = spark.readStream
.format("rate")
.option("rowsPerSecond", 10)
.load()
df.withColumn("id", (rand()*100).cast("Integer"))
.withColumn("ts", lit(1).cast("Integer"))
.withColumn("name", lit("name").cast("string"))
.withColumn("price", col("value").cast("Integer"))
.withColumn("part", date_format(col("timestamp"),"yyyyMMddHHmm"))
.drop("timestamp")
.writeStream
.format("hudi")
.options(hudiOptions)
// .trigger(Trigger.ProcessingTime("10 seconds"))
.outputMode(OutputMode.Append())
.start()
```
2. start a stream read
```
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
spark.readStream.format("org.apache.hudi").load("/tmp/hoodie/hudi40").writeStream.format("console").trigger(Trigger.ProcessingTime("1 seconds")).option("checkpointLocation", "/tmp/hudi/ckp40").start()
```
3. The read task will get the described exception later.
**Expected behavior**
A clear and concise description of what you expected to happen.
**Environment Description**
* Hudi version : master
* Spark version : 3.3.2
* Hive version :
* Hadoop version : 3.3.0
* Storage (HDFS/S3/GCS..) : hdfs
* Running on Docker? (yes/no) : no
**Additional context**
Add any other context about the problem here.
**Stacktrace**
```Add the stacktrace of the error.```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Archived parquet file lenght is 0 when spark do streaming read [hudi]
Posted by "xicm (via GitHub)" <gi...@apache.org>.
xicm commented on issue #10881:
URL: https://github.com/apache/hudi/issues/10881#issuecomment-2005676734
This is the parquet file
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Archived parquet file lenght is 0 when spark do streaming read [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on issue #10881:
URL: https://github.com/apache/hudi/issues/10881#issuecomment-2006256237
Fixed in https://github.com/apache/hudi/pull/10883/files
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Archived parquet file lenght is 0 when spark do streaming read [hudi]
Posted by "xicm (via GitHub)" <gi...@apache.org>.
xicm commented on issue #10881:
URL: https://github.com/apache/hudi/issues/10881#issuecomment-2006235716
The cause is that we update the manifest and version file before we close the parquet writer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Archived parquet file lenght is 0 when spark do streaming read [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on issue #10881:
URL: https://github.com/apache/hudi/issues/10881#issuecomment-2005697002
So you can read the parquet file with spark native reader correctly? Is there any error log from the writer while doing the archiving?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Archived parquet file lenght is 0 when spark do streaming read [hudi]
Posted by "xicm (via GitHub)" <gi...@apache.org>.
xicm closed issue #10881: [SUPPORT] Archived parquet file lenght is 0 when spark do streaming read
URL: https://github.com/apache/hudi/issues/10881
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org