Skip to content

[HUDI-9334] Optimize Parallelism of show_invalid_parquet #13206

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhangyue19921010 merged 8 commits into from
Jun 16, 2025
Merged

[HUDI-9334] Optimize Parallelism of show_invalid_parquet #13206

Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
0aef7df
[HUDI-9334] optimize parallelism of show_invalid_parquet
Apr 22, 2025
46f1daa
solve type mismatch
Apr 23, 2025
6514805
introduce customParallelism
Jun 12, 2025
1ee1c35
rm invalid imports
Jun 12, 2025
49d72c8
switch parallelism to optional
Jun 12, 2025
6662689
fix scalastyle
Jun 13, 2025
d08f745
fix scalastyle
Jun 13, 2025
c0c05f1
Merge branch 'master' into optimize-show-invalid-parquet
Jun 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 29 additions & 33 deletions ...main/scala/org/apache/spark/sql/hudi/command/procedures/ShowInvalidParquetProcedure.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@ import collection.JavaConverters._
import org.apache.hadoop.fs.Path
import org.apache.parquet.format.converter.ParquetMetadataConverter.SKIP_ROW_GROUPS
import org.apache.parquet.hadoop.ParquetFileReader
import org.apache.spark.api.java.JavaRDD
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}

Expand Down Expand Up @@ -65,42 +64,39 @@ class ShowInvalidParquetProcedure extends BaseProcedure with ProcedureBuilder {
val metadataConfig = HoodieMetadataConfig.newBuilder.enable(false).build
val metadata = HoodieTableMetadata.create(new HoodieSparkEngineContext(jsc), storage, metadataConfig, srcPath)
val partitionPaths: java.util.List[String] = metadata.getPartitionPathWithPathPrefixes(partitions.split(",").toList.asJava)
val partitionPathsSize = if (partitionPaths.size() == 0) 1 else partitionPaths.size()
val instantsList = if (StringUtils.isNullOrEmpty(instants)) Array.empty[String] else instants.split(",")
val fileStatus = partitionPaths.asScala.flatMap(part => {
val fs = HadoopFSUtils.getFs(new Path(srcPath), storageConf.unwrap())
HadoopFSUtils.getAllDataFilesInPartition(fs, HadoopFSUtils.constructAbsolutePathInHadoopPath(srcPath, part))
}).toList

val javaRdd: JavaRDD[String] = jsc.parallelize(partitionPaths, partitionPathsSize)
val parquetRdd = javaRdd.rdd.map(part => {
val fs = HadoopFSUtils.getFs(new Path(srcPath), storageConf.unwrap())
HadoopFSUtils.getAllDataFilesInPartition(fs, HadoopFSUtils.constructAbsolutePathInHadoopPath(srcPath, part)).filter(fileStatus => {
var isFilter = true
if (!instantsList.isEmpty) {
val parquetCommitTime = FSUtils.getCommitTimeWithFullPath(fileStatus.getPath.toString)
isFilter = instantsList.contains(parquetCommitTime)
}
isFilter
})
}).flatMap(_.toList)
.filter(status => {
val filePath = status.getPath
var isInvalid = false
if (filePath.toString.endsWith(".parquet")) {
try ParquetFileReader.readFooter(storageConf.unwrap(), filePath, SKIP_ROW_GROUPS).getFileMetaData catch {
case e: Exception =>
isInvalid = e.getMessage.contains("is not a Parquet file")
if (isInvalid && needDelete) {
val fs = HadoopFSUtils.getFs(new Path(srcPath), storageConf.unwrap())
try {
isInvalid = !fs.delete(filePath, false)
} catch {
case ex: Exception =>
isInvalid = true
}
val parquetRdd = jsc.parallelize(fileStatus, Math.max(fileStatus.size, 1)).filter(fileStatus => {

@TheR1sing3un TheR1sing3un Apr 23, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about allowing a custom parallelism to be passed here and then aggregating the file status into this parallelism? Using the number of partitions as the parallelism leads to too low concurrency and a long running time for a single task. But will directly using the number of files as the concurrency degree result in too many tasks? In some scenarios, tens of thousands of files are possible, but tens of thousands of concurrent degrees will put a lot of pressure on the task scheduling of spark.

@fhan688 fhan688 Apr 23, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks for advising, I'll introduce a custom parallelism for avoiding too many tasks

if (instantsList.nonEmpty) {
val parquetCommitTime = FSUtils.getCommitTimeWithFullPath(fileStatus.getPath.toString)
instantsList.contains(parquetCommitTime)
} else {
true
}}).filter(status => {
val filePath = status.getPath
var isInvalid = false
if (filePath.toString.endsWith(".parquet")) {
try ParquetFileReader.readFooter(storageConf.unwrap(), filePath, SKIP_ROW_GROUPS).getFileMetaData catch {
case e: Exception =>
isInvalid = e.getMessage.contains("is not a Parquet file")
if (isInvalid && needDelete) {
val fs = HadoopFSUtils.getFs(new Path(srcPath), storageConf.unwrap())
try {
isInvalid = !fs.delete(filePath, false)
} catch {
case ex: Exception =>
isInvalid = true
}
}
}
}
isInvalid
})
.map(status => Row(status.getPath.toString))
}
isInvalid
}).map(status => Row(status.getPath.toString))

if (limit.isDefined) {
parquetRdd.take(limit.get.asInstanceOf[Int])
} else {
Expand Down