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

Show file tree
Hide file tree
Changes from all 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: 33 additions & 29 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 All @@ -38,10 +37,11 @@ import java.util.function.Supplier
class ShowInvalidParquetProcedure extends BaseProcedure with ProcedureBuilder {
private val PARAMETERS = Array[ProcedureParameter](
ProcedureParameter.required(0, "path", DataTypes.StringType),
ProcedureParameter.optional(1, "limit", DataTypes.IntegerType, 100),
ProcedureParameter.optional(2, "needDelete", DataTypes.BooleanType, false),
ProcedureParameter.optional(3, "partitions", DataTypes.StringType, ""),
ProcedureParameter.optional(4, "instants", DataTypes.StringType, "")
ProcedureParameter.optional(1, "parallelism", DataTypes.IntegerType, 100),
ProcedureParameter.optional(2, "limit", DataTypes.IntegerType, 100),
ProcedureParameter.optional(3, "needDelete", DataTypes.BooleanType, false),
ProcedureParameter.optional(4, "partitions", DataTypes.StringType, ""),
ProcedureParameter.optional(5, "instants", DataTypes.StringType, "")
)

private val OUTPUT_TYPE = new StructType(Array[StructField](
Expand All @@ -56,31 +56,34 @@ class ShowInvalidParquetProcedure extends BaseProcedure with ProcedureBuilder {
super.checkArgs(PARAMETERS, args)

val srcPath = getArgValueOrDefault(args, PARAMETERS(0)).get.asInstanceOf[String]
val limit = getArgValueOrDefault(args, PARAMETERS(1))
val needDelete = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Boolean]
val partitions = getArgValueOrDefault(args, PARAMETERS(3)).map(_.toString).getOrElse("")
val instants = getArgValueOrDefault(args, PARAMETERS(4)).map(_.toString).getOrElse("")
val parallelism = getArgValueOrDefault(args, PARAMETERS(1)).get.asInstanceOf[Int]
val limit = getArgValueOrDefault(args, PARAMETERS(2))
val needDelete = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
val partitions = getArgValueOrDefault(args, PARAMETERS(4)).map(_.toString).getOrElse("")
val instants = getArgValueOrDefault(args, PARAMETERS(5)).map(_.toString).getOrElse("")

val storageConf = HadoopFSUtils.getStorageConfWithCopy(jsc.hadoopConfiguration())
val storage = new HoodieHadoopStorage(srcPath, storageConf)
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 => {
if (fileStatus.isEmpty) {
Seq.empty
} else {
val parquetRdd = jsc.parallelize(fileStatus, Math.min(fileStatus.size, parallelism)).filter(fileStatus => {
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")) {
Expand All @@ -99,12 +102,13 @@ class ShowInvalidParquetProcedure extends BaseProcedure with ProcedureBuilder {
}
}
isInvalid
})
.map(status => Row(status.getPath.toString))
if (limit.isDefined) {
parquetRdd.take(limit.get.asInstanceOf[Int])
} else {
parquetRdd.collect()
}).map(status => Row(status.getPath.toString))

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

Expand Down
24 changes: 14 additions & 10 deletions .../src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowInvalidParquetProcedure.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ class TestShowInvalidParquetProcedure extends HoodieSparkProcedureTestBase {
withTempDir { tmp =>
val tableName = generateTableName
val basePath = s"${tmp.getCanonicalPath}/$tableName"
val customParallelism = 3
// create table
spark.sql(
s"""
Expand Down Expand Up @@ -68,13 +69,13 @@ class TestShowInvalidParquetProcedure extends HoodieSparkProcedureTestBase {

// collect result for table
var result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism)""".stripMargin).collect()
assertResult(2) {
result.length
}

result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', limit => 1)""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, limit => 1)""".stripMargin).collect()
assertResult(1) {
result.length
}
Expand All @@ -85,6 +86,7 @@ class TestShowInvalidParquetProcedure extends HoodieSparkProcedureTestBase {
withTempDir { tmp =>
val tableName = generateTableName
val basePath = s"${tmp.getCanonicalPath}/$tableName"
val customParallelism = 3
// create table
spark.sql(
s"""
Expand Down Expand Up @@ -118,7 +120,7 @@ class TestShowInvalidParquetProcedure extends HoodieSparkProcedureTestBase {

// collect result for table
val result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', needDelete => true)""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, needDelete => true)""".stripMargin).collect()
assertResult(0) {
result.length
}
Expand All @@ -129,6 +131,7 @@ class TestShowInvalidParquetProcedure extends HoodieSparkProcedureTestBase {
withTempDir { tmp =>
val tableName = generateTableName
val basePath = s"${tmp.getCanonicalPath}/$tableName"
val customParallelism = 3
// create table
spark.sql(
s"""
Expand Down Expand Up @@ -192,46 +195,47 @@ class TestShowInvalidParquetProcedure extends HoodieSparkProcedureTestBase {

// collect result for table
var result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism)""".stripMargin).collect()
assertResult(4) {
result.length
}

result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', partitions => 'year=2022')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, partitions => 'year=2022')""".stripMargin).collect()
assertResult(4) {
result.length
}

result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', partitions => 'year=2022/month=07')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, partitions => 'year=2022/month=07')""".stripMargin).collect()
assertResult(2) {
result.length
}

result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', partitions => 'year=2022/month=08/day=30,year=2022/month=08/day=31')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, partitions => 'year=2022/month=08/day=30,year=2022/month=08/day=31')""".stripMargin).collect()
assertResult(2) {
result.length
}

result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', partitions => 'year=2023')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, partitions => 'year=2023')""".stripMargin).collect()
assertResult(0) {
result.length
}

result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', instants => '$instantTime')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, instants => '$instantTime')""".stripMargin).collect()
assertResult(4) {
result.length
}
result = spark.sql(
s"""call show_invalid_parquet(path => '$basePath', instants => '$instantTime', partitions => 'year=2022/month=08')""".stripMargin).collect()
s"""call show_invalid_parquet(path => '$basePath', parallelism => $customParallelism, instants => '$instantTime', partitions => 'year=2022/month=08')""".stripMargin).collect()
assertResult(2) {
result.length
}

}
}
}

Loading