Changes to avoid breaking in-flight queries during avro2orc conversion (#2002)

aditya1105 · web-flow · commit 542757a606f3 · 2017-07-19T04:48:00.000-07:00
diff --git a/gobblin-data-management/src/main/java/gobblin/data/management/conversion/hive/converter/AbstractAvroToOrcConverter.java b/gobblin-data-management/src/main/java/gobblin/data/management/conversion/hive/converter/AbstractAvroToOrcConverter.java
@@ -17,6 +17,7 @@
 package gobblin.data.management.conversion.hive.converter;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -39,6 +40,7 @@
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.metadata.Partition;
 import org.apache.thrift.TException;
 
@@ -135,6 +137,14 @@ public String getConfigPrefix() {
   public static final String HIVE_DATASET_DESTINATION_SKIP_SETGROUP = "hive.dataset.destination.skip.setGroup";
   public static final boolean DEFAULT_HIVE_DATASET_DESTINATION_SKIP_SETGROUP = false;
 
+  /**
+   * If the property is set to true then partition dir is overwritten,
+   * else a new time-stamped partition dir is created to avoid breaking in-flight queries
+   * Check gobblin.data.management.retention.Avro2OrcStaleDatasetCleaner to clean stale directories
+   */
+  public static final String HIVE_DATASET_PARTITION_OVERWRITE = "hive.dataset.partition.overwrite";
+  public static final boolean DEFAULT_HIVE_DATASET_PARTITION_OVERWRITE = true;
+
   /**
    * If set to true, a set format DDL will be separate from add partition DDL
    */
@@ -459,31 +469,35 @@ public Iterable<QueryBasedHiveConversionEntity> convertRecord(Schema outputAvroS
       // Step:
       // A.2.3, B.2.3: If partitioned table, move partitions from staging to final table; for all partitions:
 
-      // Step:
-      // A.2.3.1, B.2.3.1: Drop if exists partition in final table
-      List<String> dropPartitionsDDL =
-          HiveAvroORCQueryGenerator.generateDropPartitionsDDL(orcTableDatabase,
-              orcTableName,
-              partitionsDMLInfo);
-      log.debug("Drop partitions if exist in final table: " + dropPartitionsDDL);
-      publishQueries.addAll(dropPartitionsDDL);
-
       // Step:
       // A.2.3.2, B.2.3.2: Move partition directory
       // Move: orcStagingDataPartitionLocation to: orcFinalDataPartitionLocation
       String orcFinalDataPartitionLocation = orcDataLocation + Path.SEPARATOR + orcStagingDataPartitionDirName;
-      log.info("Partition directory to move: " + orcStagingDataPartitionLocation + " to: " + orcFinalDataPartitionLocation);
+      Optional<Path> destPartitionLocation = getDestinationPartitionLocation(destinationTableMeta, workUnit,
+          conversionEntity.getHivePartition().get().getName());
+      orcFinalDataPartitionLocation =
+          updatePartitionLocation(orcFinalDataPartitionLocation, workUnit, destPartitionLocation);
+      log.info(
+          "Partition directory to move: " + orcStagingDataPartitionLocation + " to: " + orcFinalDataPartitionLocation);
       publishDirectories.put(orcStagingDataPartitionLocation, orcFinalDataPartitionLocation);
+      // Step:
+      // A.2.3.1, B.2.3.1: Drop if exists partition in final table
 
       // Step:
+      // If destination partition already exists, alter the partition location
       // A.2.3.3, B.2.3.3: Create partition with location (and update storage format if not in ORC already)
-      String orcDataPartitionLocation = orcDataLocation + Path.SEPARATOR + orcStagingDataPartitionDirName;
+      List<String> dropPartitionsDDL =
+          HiveAvroORCQueryGenerator.generateDropPartitionsDDL(orcTableDatabase,
+              orcTableName,
+              partitionsDMLInfo);
+      log.debug("Drop partitions if exist in final table: " + dropPartitionsDDL);
+      publishQueries.addAll(dropPartitionsDDL);
       if (workUnit.getPropAsBoolean(HIVE_CONVERSION_SETSERDETOAVROEXPLICITELY,
           DEFAULT_HIVE_CONVERSION_SETSERDETOAVROEXPLICITELY)) {
         List<String> createFinalPartitionDDL =
             HiveAvroORCQueryGenerator.generateCreatePartitionDDL(orcTableDatabase,
                 orcTableName,
-                orcDataPartitionLocation,
+                orcFinalDataPartitionLocation,
                 partitionsDMLInfo,
                 Optional.<String>absent());
 
@@ -503,7 +517,7 @@ public Iterable<QueryBasedHiveConversionEntity> convertRecord(Schema outputAvroS
         List<String> createFinalPartitionDDL =
             HiveAvroORCQueryGenerator.generateCreatePartitionDDL(orcTableDatabase,
                 orcTableName,
-                orcDataPartitionLocation,
+                orcFinalDataPartitionLocation,
                 partitionsDMLInfo,
                 Optional.fromNullable(ORC_FORMAT));
 
@@ -747,4 +761,50 @@ private Pair<Optional<Table>, Optional<List<Partition>>> getDestinationTableMeta
 
     return ImmutablePair.of(table, partitions);
   }
+
+  /**
+   * If partition already exists then new partition location will be a separate time stamp dir
+   * If partition location is /a/b/c/<oldTimeStamp> then new partition location is /a/b/c/<currentTimeStamp>
+   * If partition location is /a/b/c/ then new partition location is /a/b/c/<currentTimeStamp>
+   **/
+  private String updatePartitionLocation(String orcDataPartitionLocation, WorkUnitState workUnitState,
+      Optional<Path> destPartitionLocation)
+      throws DataConversionException {
+
+    if (workUnitState.getPropAsBoolean(HIVE_DATASET_PARTITION_OVERWRITE, DEFAULT_HIVE_DATASET_PARTITION_OVERWRITE)) {
+      return orcDataPartitionLocation;
+    }
+    if (!destPartitionLocation.isPresent()) {
+      return orcDataPartitionLocation;
+    }
+    long timeStamp = System.currentTimeMillis();
+    return StringUtils.join(Arrays.asList(orcDataPartitionLocation, timeStamp), '/');
+  }
+
+  private Optional<Path> getDestinationPartitionLocation(Optional<Table> table, WorkUnitState state,
+      String partitionName)
+      throws DataConversionException {
+    Optional<org.apache.hadoop.hive.metastore.api.Partition> partitionOptional =
+        Optional.<org.apache.hadoop.hive.metastore.api.Partition>absent();
+    if (!table.isPresent()) {
+      return Optional.<Path>absent();
+    }
+    try {
+      HiveMetastoreClientPool pool = HiveMetastoreClientPool.get(state.getJobState().getProperties(),
+          Optional.fromNullable(state.getJobState().getProp(HiveDatasetFinder.HIVE_METASTORE_URI_KEY)));
+      try (AutoReturnableObject<IMetaStoreClient> client = pool.getClient()) {
+        partitionOptional =
+            Optional.of(client.get().getPartition(table.get().getDbName(), table.get().getTableName(), partitionName));
+      }
+      if (partitionOptional.isPresent()) {
+        org.apache.hadoop.hive.ql.metadata.Table qlTable = new org.apache.hadoop.hive.ql.metadata.Table(table.get());
+        org.apache.hadoop.hive.ql.metadata.Partition qlPartition =
+            new org.apache.hadoop.hive.ql.metadata.Partition(qlTable, partitionOptional.get());
+        return Optional.of(qlPartition.getDataLocation());
+      }
+    } catch (IOException | TException | HiveException e) {
+      throw new DataConversionException("Could not fetch destination table metadata", e);
+    }
+    return Optional.<Path>absent();
+  }
 }
diff --git a/gobblin-modules/gobblin-azkaban/src/main/java/gobblin/data/management/retention/Avro2OrcStaleDatasetCleaner.java b/gobblin-modules/gobblin-azkaban/src/main/java/gobblin/data/management/retention/Avro2OrcStaleDatasetCleaner.java
@@ -0,0 +1,128 @@
+package gobblin.data.management.retention;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Optional;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import azkaban.jobExecutor.AbstractJob;
+
+import gobblin.data.management.conversion.hive.dataset.ConvertibleHiveDataset;
+import gobblin.data.management.conversion.hive.dataset.ConvertibleHiveDatasetFinder;
+import gobblin.data.management.conversion.hive.events.EventConstants;
+import gobblin.data.management.conversion.hive.validation.ValidationJob;
+import gobblin.data.management.copy.hive.HiveDataset;
+import gobblin.data.management.copy.hive.HiveDatasetFinder;
+import gobblin.data.management.copy.hive.HiveUtils;
+import gobblin.instrumented.Instrumented;
+import gobblin.metrics.MetricContext;
+import gobblin.metrics.event.EventSubmitter;
+import gobblin.util.AutoReturnableObject;
+import gobblin.util.ConfigUtils;
+
+
+public class Avro2OrcStaleDatasetCleaner extends AbstractJob {
+  private static final Logger log = Logger.getLogger(ValidationJob.class);
+  private static final String HIVE_PARTITION_DELETION_GRACE_TIME_IN_DAYS = "hive.partition.deletion.graceTime.inDays";
+  private static final String DEFAULT_HIVE_PARTITION_DELETION_GRACE_TIME_IN_DAYS = "2";
+  private final MetricContext metricContext;
+  private final EventSubmitter eventSubmitter;
+  private final ConvertibleHiveDatasetFinder datasetFinder;
+  private static final String HIVE_DATASET_CONFIG_AVRO_PREFIX = "hive.conversion.avro";
+  private final FileSystem fs;
+  private final long graceTimeInMillis;
+
+  public Avro2OrcStaleDatasetCleaner(String jobId, Properties props)
+      throws IOException {
+    super(jobId, log);
+    props.setProperty(HiveDatasetFinder.HIVE_DATASET_CONFIG_PREFIX_KEY, HIVE_DATASET_CONFIG_AVRO_PREFIX);
+    this.graceTimeInMillis = TimeUnit.DAYS.toMillis(Long.parseLong(props
+        .getProperty(HIVE_PARTITION_DELETION_GRACE_TIME_IN_DAYS, DEFAULT_HIVE_PARTITION_DELETION_GRACE_TIME_IN_DAYS)));
+    Config config = ConfigFactory.parseProperties(props);
+    this.fs = FileSystem.newInstance(new Configuration());
+    this.metricContext = Instrumented.getMetricContext(ConfigUtils.configToState(config), ValidationJob.class);
+    this.eventSubmitter = new EventSubmitter.Builder(this.metricContext, EventConstants.CONVERSION_NAMESPACE).build();
+    this.datasetFinder = new ConvertibleHiveDatasetFinder(this.fs, props, this.eventSubmitter);
+  }
+
+  @Override
+  public void run()
+      throws Exception {
+    Iterator<HiveDataset> iterator = this.datasetFinder.getDatasetsIterator();
+    while (iterator.hasNext()) {
+      ConvertibleHiveDataset hiveDataset = (ConvertibleHiveDataset) iterator.next();
+      try (AutoReturnableObject<IMetaStoreClient> client = hiveDataset.getClientPool().getClient()) {
+        Set<Partition> sourcePartitions =
+            new HashSet<>(HiveUtils.getPartitions(client.get(), hiveDataset.getTable(), Optional.<String>absent()));
+
+        sourcePartitions.parallelStream().filter(partition -> isUnixTimeStamp(partition.getDataLocation().getName()))
+            .forEach(partition -> {
+              Arrays.stream(listFiles(partition.getDataLocation().getParent())).filter(
+                  fileStatus -> !fileStatus.getPath().toString()
+                      .equalsIgnoreCase(partition.getDataLocation().toString())).forEach(fileStatus -> {
+                deletePath(fileStatus, this.graceTimeInMillis, true);
+              });
+            });
+      }
+    }
+  }
+
+  private FileStatus[] listFiles(Path path) {
+    try {
+      return this.fs.listStatus(path);
+    } catch (IOException e) {
+      log.error("Unalbe to list files for directory " + path, e);
+      return new FileStatus[0];
+    }
+  }
+
+  private void deletePath(FileStatus fileStatus, long graceTimeInMillis, boolean recursively) {
+    long modificationTime = fileStatus.getModificationTime();
+    long currentTime = System.currentTimeMillis();
+    if ((currentTime - modificationTime) < 0) {
+      log.error("Modification time cannot be greater than current time: " + fileStatus.getPath());
+      return;
+    }
+    if ((currentTime - modificationTime) < graceTimeInMillis) {
+      log.info("Modification time is still within grace time for deletion: " + fileStatus.getPath());
+      return;
+    }
+    try {
+      this.fs.delete(fileStatus.getPath(), recursively);
+      log.info("Deleted path " + fileStatus.getPath());
+    } catch (IOException e) {
+      log.error("Unable to delete directory " + fileStatus.getPath(), e);
+    }
+  }
+
+  /**
+   * Check if a given string is a valid unixTimeStamp
+   */
+  private static boolean isUnixTimeStamp(String timeStamp) {
+    int TIME_STAMP_LENGTH = 13;
+    if (timeStamp.length() != TIME_STAMP_LENGTH) {
+      return false;
+    }
+    try {
+      Long.parseLong(timeStamp);
+      return true;
+    } catch (NumberFormatException e) {
+      return false;
+    }
+  }
+}
