You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2020/06/27 01:29:20 UTC

[GitHub] [hudi] umehrot2 opened a new pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

umehrot2 opened a new pull request #1768:
URL: https://github.com/apache/hudi/pull/1768


   ## *Tips*
   - *Thank you very much for contributing to Apache Hudi.*
   - *Please review https://hudi.apache.org/contributing.html before opening a pull request.*
   
   ## What is the purpose of the pull request
   
   This PR does several performance improvements described in https://issues.apache.org/jira/browse/HUDI-1054 that are specially useful for S3, but will be beneficial in general for hudi performance too.
   
   ## Brief change log
   
   *(for example:)*
     - *Modify AnnotationLocation checkstyle rule in checkstyle.xml*
   
   ## Verify this pull request
   
   *(Please pick either of the following options)*
   
   This pull request is a trivial rework / code cleanup without any test coverage.
   
   *(or)*
   
   This pull request is already covered by existing tests, such as *(please describe tests)*.
   
   (or)
   
   This change added tests and can be verified as follows:
   
   *(example:)*
   
     - *Added integration tests for end-to-end.*
     - *Added HoodieClientWriteTest to verify the change.*
     - *Manually verified the change by running a job locally.*
   
   ## Committer checklist
   
    - [ ] Has a corresponding JIRA in PR title & commit
    
    - [ ] Commit message is descriptive of the change
    
    - [ ] CI is green
   
    - [ ] Necessary doc changes done or have another open PR
          
    - [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r447595376



##########
File path: hudi-common/pom.xml
##########
@@ -147,6 +147,16 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- Spark -->
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql_${scala.binary.version}</artifactId>
+    </dependency>
+

Review comment:
       yes.. that ll do.. 




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-666767040


   > LGTM overall.
   > 
   > One high level question though. this parallelizes till the first level only right? so we are assuming this helps the common cases like date based tables with multiple years of data? I mean - if you only have a few years of data <10 say, and `yyyy` is the top level partitioning field, would this parallelization still help?
   
   @vinothchandar you are right about this. It will parallelize only on the top level partition folder. I think this will still help with parallelization, and would work best where there is only one level of partitioning. But I agree there is scope to further improve this by getting leaf level partition directories instead to help with multi level partitioning scenario. Is it okay if I open a JIRA for this and pursue it separately ?


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r455390259



##########
File path: hudi-client/src/main/java/org/apache/hudi/table/HoodieTable.java
##########
@@ -386,13 +389,26 @@ public void finalizeWrite(JavaSparkContext jsc, String instantTs, List<HoodieWri
    *
    * @param instantTs Instant Time
    */
-  public void deleteMarkerDir(String instantTs) {
+  public void deleteMarkerDir(JavaSparkContext jsc, String instantTs) {
     try {
       FileSystem fs = getMetaClient().getFs();
       Path markerDir = new Path(metaClient.getMarkerFolderPath(instantTs));
+
       if (fs.exists(markerDir)) {
         // For append only case, we do not write to marker dir. Hence, the above check
-        LOG.info("Removing marker directory=" + markerDir);
+        LOG.info("Removing marker directory = " + markerDir);
+
+        FileStatus[] fileStatuses = fs.listStatus(markerDir);

Review comment:
       @vinothchandar I don't think there is any different between `EmrFS which uses S3` vs `HDFS` working w.r.t RPC calls made here. EmrFS just implements the HDFS interface, but the internals like RPC calls to the namenode etc. remain the same.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r463323604



##########
File path: hudi-client/src/main/java/org/apache/hudi/table/MarkerFiles.java
##########
@@ -110,6 +158,10 @@ private String translateMarkerToDataPath(String markerPath) {
     return MarkerFiles.stripMarkerSuffix(rPath);
   }
 
+  public static String stripMarkerSuffix(String path) {

Review comment:
       I can change it back, but is there a code style guide we are following ? I haven't seen this rule in google style guide for java atleast.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-650479862


   @bvaradar fyi


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r447264262



##########
File path: hudi-common/pom.xml
##########
@@ -147,6 +147,16 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- Spark -->
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql_${scala.binary.version}</artifactId>
+    </dependency>
+

Review comment:
       Okay so is the suggestion to move this method over to `HoodieTable` in `hudi-client` module, considering that is the only place this method is used ?




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] zhangyue19921010 commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
zhangyue19921010 commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-964965267


   Hi guys, this PR helps a lot to do archive works based on S3!  Also Just find an another performance issues and raise a PR trying to fix it. https://github.com/apache/hudi/pull/3920 
   Hope you are interested in it :)


-- 
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

[GitHub] [hudi] umehrot2 commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-664741636


   > @umehrot2 just landed the changes I mentioned. can we rework this PR and try again . We can make things parallel i.e working for s3 for now. and then we can adjust for HDFS later on. So we should be able close the loop faster.
   > 
   > I do want to get this into 0.6.0 so also please let me know if you are unable to take a stab at this
   
   Working on it @vinothchandar. There has been quite a refactoring it seems, which is making the re-basing tricky as now these functions are being called from places which do not even have `spark context`.


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-661679756


   @umehrot2 just landed the changes I mentioned. can we rework this PR and try again . We can make things parallel i.e working for s3 for now. and then we can adjust for HDFS later on. So we should be able close the loop faster.
   
   I do want to get this into 0.6.0 so also please let me know if you are unable to take a stab at this 


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar edited a comment on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar edited a comment on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-666919866


   > Is it okay if I open a JIRA for this and pursue it separately ?
   
   Sounds good. let's lump this into the JIRA we have for marker file improvements more holistically ?


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r463404703



##########
File path: hudi-client/src/main/java/org/apache/hudi/table/MarkerFiles.java
##########
@@ -110,6 +158,10 @@ private String translateMarkerToDataPath(String markerPath) {
     return MarkerFiles.stripMarkerSuffix(rPath);
   }
 
+  public static String stripMarkerSuffix(String path) {

Review comment:
       this is not checkstyle enforced atm. So I am happy to let this be. 




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-665999807


   @vinothchandar finally got the unit and integration tests to pass. This is ready for review.


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-651720918


   marked this a blocker for 0.6.0.. @n3nash can you please chime in with any side effects you see for hdfs?


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r462994647



##########
File path: hudi-client/src/main/java/org/apache/hudi/table/MarkerFiles.java
##########
@@ -64,44 +69,87 @@ public MarkerFiles(HoodieTable<?> table, String instantTime) {
         instantTime);
   }
 
-  public void quietDeleteMarkerDir() {
+  public void quietDeleteMarkerDir(JavaSparkContext jsc, int parallelism) {
     try {
-      deleteMarkerDir();
+      deleteMarkerDir(jsc, parallelism);
     } catch (HoodieIOException ioe) {
       LOG.warn("Error deleting marker directory for instant " + instantTime, ioe);
     }
   }
 
   /**
    * Delete Marker directory corresponding to an instant.
+   *
+   * @param jsc Java Spark Context
+   * @param parallelism Spark parallelism for deletion
    */
-  public boolean deleteMarkerDir() {
+  public boolean deleteMarkerDir(JavaSparkContext jsc, int parallelism) {
     try {
-      boolean result = fs.delete(markerDirPath, true);
-      if (result) {
+      if (fs.exists(markerDirPath)) {
+        FileStatus[] fileStatuses = fs.listStatus(markerDirPath);
+        List<String> markerDirSubPaths = Arrays.stream(fileStatuses)
+                .map(fileStatus -> fileStatus.getPath().toString())
+                .collect(Collectors.toList());
+
+        if (markerDirSubPaths.size() > 0) {
+          SerializableConfiguration conf = new SerializableConfiguration(fs.getConf());
+          parallelism = Math.min(markerDirSubPaths.size(), parallelism);
+          jsc.parallelize(markerDirSubPaths, parallelism).foreach(subPathStr -> {
+            Path subPath = new Path(subPathStr);
+            FileSystem fileSystem = subPath.getFileSystem(conf.get());
+            fileSystem.delete(subPath, true);

Review comment:
       note to self: this will still work when `subPath` is a file. i.e non-partitioned tables

##########
File path: hudi-client/src/main/java/org/apache/hudi/table/MarkerFiles.java
##########
@@ -110,6 +158,10 @@ private String translateMarkerToDataPath(String markerPath) {
     return MarkerFiles.stripMarkerSuffix(rPath);
   }
 
+  public static String stripMarkerSuffix(String path) {

Review comment:
       nit: better to have the static methods right at the top?




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r447264716



##########
File path: hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
##########
@@ -199,16 +201,40 @@ public static String getRelativePartitionPath(Path basePath, Path partitionPath)
     return partitions;
   }
 
-  public static List<String> getAllDataFilesForMarkers(FileSystem fs, String basePath, String instantTs,
-      String markerDir, String baseFileExtension) throws IOException {
-    List<String> dataFiles = new LinkedList<>();
-    processFiles(fs, markerDir, (status) -> {
-      String pathStr = status.getPath().toString();
-      if (pathStr.endsWith(HoodieTableMetaClient.MARKER_EXTN)) {
-        dataFiles.add(FSUtils.translateMarkerToDataPath(basePath, pathStr, instantTs, baseFileExtension));
+  public static Set<String> getAllDataFilesForMarkers(JavaSparkContext jsc, FileSystem fs, String basePath,
+      String instantTs, String markerDir, String baseFileExtension, int parallelism) throws IOException {
+    FileStatus[] topLevelStatuses = fs.listStatus(new Path(markerDir));
+
+    Set<String> dataFiles = new HashSet<>();
+
+    List<String> subDirectories = new ArrayList<>();
+    for (FileStatus topLevelStatus: topLevelStatuses) {
+      if (topLevelStatus.isFile()) {
+        String pathStr = topLevelStatus.getPath().toString();
+        if (pathStr.endsWith(HoodieTableMetaClient.MARKER_EXTN)) {
+          dataFiles.add(FSUtils.translateMarkerToDataPath(basePath, pathStr, instantTs, baseFileExtension));
+        }
+      } else {
+        subDirectories.add(topLevelStatus.getPath().toString());
       }
-      return true;
-    }, false);
+    }
+
+    parallelism = subDirectories.size() < parallelism ? subDirectories.size() : parallelism;

Review comment:
       Will address in the next revision, one I have more feedback as needed from @n3nash 




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar merged pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar merged pull request #1768:
URL: https://github.com/apache/hudi/pull/1768


   


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-666919866


   > Is it okay if I open a JIRA for this and pursue it separately ?
   Sounds good. let's lump this into the JIRA we have for marker file improvements more holistically ?


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] leesf commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
leesf commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r446502715



##########
File path: hudi-common/pom.xml
##########
@@ -147,6 +147,16 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- Spark -->
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql_${scala.binary.version}</artifactId>
+    </dependency>
+

Review comment:
       `hudi-common` is a base module and feels a little weird that it relies on spark-core/spark-sql, would we remove them and move `getAllDataFilesForMarkers` method to `hudi-client` module, wdyt? cc @bvaradar @vinothchandar 




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] umehrot2 commented on pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
umehrot2 commented on pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#issuecomment-667241386


   > > Is it okay if I open a JIRA for this and pursue it separately ?
   > 
   > Sounds good. let's lump this into the JIRA we have for marker file improvements more holistically ?
   
   Added a comment about this on https://issues.apache.org/jira/browse/HUDI-1138. Let me try to fix some conflict that it is showing with current master.


----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on a change in pull request #1768: [HUDI-1054][Peformance] Several performance fixes during finalizing writes

Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #1768:
URL: https://github.com/apache/hudi/pull/1768#discussion_r446621351



##########
File path: hudi-common/pom.xml
##########
@@ -147,6 +147,16 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- Spark -->
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql_${scala.binary.version}</artifactId>
+    </dependency>
+

Review comment:
       yes.. we cannot depend on spark in `hudi-common` 

##########
File path: hudi-client/src/main/java/org/apache/hudi/table/HoodieTable.java
##########
@@ -386,13 +389,26 @@ public void finalizeWrite(JavaSparkContext jsc, String instantTs, List<HoodieWri
    *
    * @param instantTs Instant Time
    */
-  public void deleteMarkerDir(String instantTs) {
+  public void deleteMarkerDir(JavaSparkContext jsc, String instantTs) {
     try {
       FileSystem fs = getMetaClient().getFs();
       Path markerDir = new Path(metaClient.getMarkerFolderPath(instantTs));
+
       if (fs.exists(markerDir)) {
         // For append only case, we do not write to marker dir. Hence, the above check
-        LOG.info("Removing marker directory=" + markerDir);
+        LOG.info("Removing marker directory = " + markerDir);
+
+        FileStatus[] fileStatuses = fs.listStatus(markerDir);

Review comment:
       cc @n3nash should we have flag to protect this for HDFS.. i.e if the recursive delete works better there (IIUC). you might want to tradeoff less RPCs ..? 
   we can override defaults at spark datasource level, and set these based on `StorageSchemes` as well. 

##########
File path: hudi-client/src/main/java/org/apache/hudi/table/HoodieTable.java
##########
@@ -386,13 +389,26 @@ public void finalizeWrite(JavaSparkContext jsc, String instantTs, List<HoodieWri
    *
    * @param instantTs Instant Time
    */
-  public void deleteMarkerDir(String instantTs) {
+  public void deleteMarkerDir(JavaSparkContext jsc, String instantTs) {
     try {
       FileSystem fs = getMetaClient().getFs();
       Path markerDir = new Path(metaClient.getMarkerFolderPath(instantTs));
+
       if (fs.exists(markerDir)) {
         // For append only case, we do not write to marker dir. Hence, the above check
-        LOG.info("Removing marker directory=" + markerDir);
+        LOG.info("Removing marker directory = " + markerDir);
+
+        FileStatus[] fileStatuses = fs.listStatus(markerDir);

Review comment:
       @umehrot2 so seems like, for object stores this is different..  and makes sense completely to do parallel cleaning of individual files.

##########
File path: hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
##########
@@ -199,16 +201,40 @@ public static String getRelativePartitionPath(Path basePath, Path partitionPath)
     return partitions;
   }
 
-  public static List<String> getAllDataFilesForMarkers(FileSystem fs, String basePath, String instantTs,
-      String markerDir, String baseFileExtension) throws IOException {
-    List<String> dataFiles = new LinkedList<>();
-    processFiles(fs, markerDir, (status) -> {
-      String pathStr = status.getPath().toString();
-      if (pathStr.endsWith(HoodieTableMetaClient.MARKER_EXTN)) {
-        dataFiles.add(FSUtils.translateMarkerToDataPath(basePath, pathStr, instantTs, baseFileExtension));
+  public static Set<String> getAllDataFilesForMarkers(JavaSparkContext jsc, FileSystem fs, String basePath,

Review comment:
       i think this is the reason for needing spark in `hudi-common`.. we can move refactor the code to `hudi-client`.. 
   
   In fact, #1755  has already modularized this more.. 

##########
File path: hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
##########
@@ -199,16 +201,40 @@ public static String getRelativePartitionPath(Path basePath, Path partitionPath)
     return partitions;
   }
 
-  public static List<String> getAllDataFilesForMarkers(FileSystem fs, String basePath, String instantTs,
-      String markerDir, String baseFileExtension) throws IOException {
-    List<String> dataFiles = new LinkedList<>();
-    processFiles(fs, markerDir, (status) -> {
-      String pathStr = status.getPath().toString();
-      if (pathStr.endsWith(HoodieTableMetaClient.MARKER_EXTN)) {
-        dataFiles.add(FSUtils.translateMarkerToDataPath(basePath, pathStr, instantTs, baseFileExtension));
+  public static Set<String> getAllDataFilesForMarkers(JavaSparkContext jsc, FileSystem fs, String basePath,
+      String instantTs, String markerDir, String baseFileExtension, int parallelism) throws IOException {
+    FileStatus[] topLevelStatuses = fs.listStatus(new Path(markerDir));
+
+    Set<String> dataFiles = new HashSet<>();
+
+    List<String> subDirectories = new ArrayList<>();
+    for (FileStatus topLevelStatus: topLevelStatuses) {
+      if (topLevelStatus.isFile()) {
+        String pathStr = topLevelStatus.getPath().toString();
+        if (pathStr.endsWith(HoodieTableMetaClient.MARKER_EXTN)) {
+          dataFiles.add(FSUtils.translateMarkerToDataPath(basePath, pathStr, instantTs, baseFileExtension));
+        }
+      } else {
+        subDirectories.add(topLevelStatus.getPath().toString());
       }
-      return true;
-    }, false);
+    }
+
+    parallelism = subDirectories.size() < parallelism ? subDirectories.size() : parallelism;

Review comment:
       Math.min(subDirectories.size(), parallelism)?

##########
File path: hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
##########
@@ -199,16 +201,40 @@ public static String getRelativePartitionPath(Path basePath, Path partitionPath)
     return partitions;
   }
 
-  public static List<String> getAllDataFilesForMarkers(FileSystem fs, String basePath, String instantTs,
-      String markerDir, String baseFileExtension) throws IOException {
-    List<String> dataFiles = new LinkedList<>();
-    processFiles(fs, markerDir, (status) -> {
-      String pathStr = status.getPath().toString();
-      if (pathStr.endsWith(HoodieTableMetaClient.MARKER_EXTN)) {
-        dataFiles.add(FSUtils.translateMarkerToDataPath(basePath, pathStr, instantTs, baseFileExtension));
+  public static Set<String> getAllDataFilesForMarkers(JavaSparkContext jsc, FileSystem fs, String basePath,
+      String instantTs, String markerDir, String baseFileExtension, int parallelism) throws IOException {
+    FileStatus[] topLevelStatuses = fs.listStatus(new Path(markerDir));
+
+    Set<String> dataFiles = new HashSet<>();
+
+    List<String> subDirectories = new ArrayList<>();
+    for (FileStatus topLevelStatus: topLevelStatuses) {
+      if (topLevelStatus.isFile()) {
+        String pathStr = topLevelStatus.getPath().toString();
+        if (pathStr.endsWith(HoodieTableMetaClient.MARKER_EXTN)) {
+          dataFiles.add(FSUtils.translateMarkerToDataPath(basePath, pathStr, instantTs, baseFileExtension));
+        }
+      } else {
+        subDirectories.add(topLevelStatus.getPath().toString());
       }
-      return true;
-    }, false);
+    }
+
+    parallelism = subDirectories.size() < parallelism ? subDirectories.size() : parallelism;
+    dataFiles.addAll(jsc.parallelize(subDirectories, parallelism).flatMap(directory -> {

Review comment:
       similar question here.. cc @n3nash .. 




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org