[GitHub] [gobblin] autumnust commented on a change in pull request #3158: [GOBBLIN-1377] Adds functionality to create shards for target directories in hive distcp

[GitBox <gi...@apache.org>]

autumnust commented on a change in pull request #3158:
URL: https://github.com/apache/gobblin/pull/3158#discussion_r592113869



##########
File path: gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/publisher/CopyDataPublisher.java
##########
@@ -79,7 +79,7 @@
 @Slf4j
 public class CopyDataPublisher extends DataPublisher implements UnpublishedHandling {
 
-  private final Path writerOutputDir;
+  private Path writerOutputDir;

Review comment:
       if adopting suggestion below this can stay as final I guess? 

##########
File path: gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/publisher/CopyDataPublisher.java
##########
@@ -225,8 +226,13 @@ private void publishFileSet(CopyEntity.DatasetAndPartition datasetAndPartition,
     Preconditions.checkArgument(!datasetWorkUnitStates.isEmpty(),
         "publishFileSet received an empty collection work units. This is an error in code.");
 
-    CopyableDatasetMetadata metadata = CopyableDatasetMetadata
-        .deserialize(datasetWorkUnitStates.iterator().next().getProp(CopySource.SERIALIZED_COPYABLE_DATASET));
+    CopyableDatasetMetadata metadata = CopyableDatasetMetadata.deserialize(
+        datasetWorkUnitStates.iterator().next().getProp(CopySource.SERIALIZED_COPYABLE_DATASET));
+
+    // Get the dataset specific sharded directory if configured
+    if (state.getPropAsBoolean(ConfigurationKeys.USE_DATASET_LOCAL_WORK_DIR)) {
+      this.writerOutputDir = new Path(datasetWorkUnitStates.iterator().next().getProp(ConfigurationKeys.WRITER_OUTPUT_DIR));
+    }
     Path datasetWriterOutputPath = new Path(this.writerOutputDir, datasetAndPartition.identifier());

Review comment:
       1. Can we store `datasetWorkUnitStates.iterator().next()` in a temp variable like `sampledWUS` to avoid calling twice? 
   2. `writerOutputDir` in this case should be a local variable since it will be specific to a workunitstate. Maybe converting the whole thing including line 236 into a ternary operator like `Path datasetWriterOutputPath = state.getPropAsBoolean(ConfigurationKeys.USE_DATASET_LOCAL_WORK_DIR) ? xxx : this.writerOutputDir` ? just leave the `writerOutputDir` as null otherwise might be safer: we can risk fail the job but not writing something that are not supposed to be in a dataset folder
   

##########
File path: gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/hive/HiveCopyEntityHelper.java
##########
@@ -142,6 +145,13 @@
    */
   public static final String HIVE_PARTITION_EXTENDED_FILTER_TYPE = HiveDatasetFinder.HIVE_DATASET_PREFIX + ".extendedFilterType";
 
+
+  /**
+   * Config to specify class for ensuring that the target directory exists, generally for cloud providers
+   * */
+  public static final String SHARD_CLIENT_CLASS
+      = HiveDatasetFinder.HIVE_DATASET_PREFIX + ".copy.target.shardClientClass";

Review comment:
       Should this be a more generic configuration as shard-client won't only be applied to hive-copy use cases? 

##########
File path: gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/CopySource.java
##########
@@ -370,9 +366,25 @@ public Void call() {
               workUnit.setProp(ConfigurationKeys.COPY_EXPECTED_SCHEMA, ((ConfigBasedDataset) this.copyableDataset).getExpectedSchema());
             }
           }
-          if ((this.copyableDataset instanceof HiveDataset) && (state.getPropAsBoolean(ConfigurationKeys.IS_DATASET_STAGING_DIR_USED,false))) {
-            workUnit.setProp(DATASET_STAGING_DIR_PATH, ((HiveDataset) this.copyableDataset).getProperties().getProperty(DATASET_STAGING_PATH));
+
+          // Ensure that the writer temporary directories are contained within the dataset shard
+          if ((this.copyableDataset instanceof HiveDataset) && (state.getPropAsBoolean(ConfigurationKeys.USE_DATASET_LOCAL_WORK_DIR,false))) {

Review comment:
       It is OK to do it temporarily but using `instanceOf` in `CopySource` definitely seems like a abstraction leaking ( I know it was there already). 
   
   We don't have to address this in this PR, but do you think we should put something like `getDatasetPath` as part of `CopyableDatasetBase` interface?  So that we don't need to explicitly cast them

##########
File path: gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/hive/ShardDirectoryClient.java
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.data.management.copy.hive;

Review comment:
       similar comment: maybe not specific to hive copy 




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