prashantwason commented on code in PR #20211:
URL: https://github.com/apache/hudi/pull/20211#discussion_r4199277191
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean
||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from
storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across
executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud
storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ filePaths: java.util.List[StoragePathInfo]):
Boolean = {
+ val sc = sqlContext.sparkContext
+ val allFilesToCheck = filePaths.asScala.map(_.getPath.toString)
+ val storageConf =
HadoopFSUtils.getStorageConfWithCopy(sc.hadoopConfiguration)
Review Comment:
Done in 47c4fd1cb40d. `checkIfAnyFilesMissing` now ships
`metaClient.getStorageConf` to the executors instead of
`sc.hadoopConfiguration`, so session-scoped `fs.*` settings are honored.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean
||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from
storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across
executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud
storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
Review Comment:
Done in 47c4fd1cb40d. `IncrementalRelationV1` and `IncrementalRelationV2`
now call `HoodieBaseRelation.checkIfAnyFilesMissing`, so they get the capped
parallelism, the `mapPartitions` change and the meta client storage conf. The
helper now takes path strings, so it works for both the MOR relations and these.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean
||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from
storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across
executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud
storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ filePaths: java.util.List[StoragePathInfo]):
Boolean = {
+ val sc = sqlContext.sparkContext
+ val allFilesToCheck = filePaths.asScala.map(_.getPath.toString)
+ val storageConf =
HadoopFSUtils.getStorageConfWithCopy(sc.hadoopConfiguration)
+ val localBasePathStr = metaClient.getBasePath.toString
+ val numPartitions = Math.max(1, Math.min(allFilesToCheck.size,
sc.defaultParallelism))
+ val previousJobDescription = sc.getLocalProperty("spark.job.description")
+ try {
+ sc.setJobDescription(s"Checking existence of ${allFilesToCheck.size}
files for incremental fallback scan")
+ val missingFileFound = sc.parallelize(allFilesToCheck.toSeq,
numPartitions)
+ .map(path => {
+ val storage = HoodieStorageUtils.getStorage(localBasePathStr,
storageConf)
Review Comment:
Done in 47c4fd1cb40d. Switched to `mapPartitions`: storage is created once
per Spark partition, and each partition stops at the first missing file
(`forall`).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
