Modified to use the existing DiffPartitionsFromFile

Author: faizalrub-datastax

src/main/java/com/datastax/cdm/job/SplitPartitions.java

@@ -158,40 +158,4 @@ private static BufferedReader getfileReader(String fileName) {
         }
     }
 
-    public static List<Partition> getFailedSubPartitionsFromFile(int splitSize, String tokenRangeFile) throws IOException {
-        logger.info("ThreadID: {} Splitting partitions in file: {} using a split-size of {}"
-                , Thread.currentThread().getId(), tokenRangeFile, splitSize);
-
-        File file = new File(tokenRangeFile);
-        String renamedFile = tokenRangeFile+"_bkp";
-        File rename = new File(renamedFile);
-        if(rename.exists()) {
-            rename.delete();
-        }
-        boolean flag = file.renameTo(rename);
-        if (flag) {
-            logger.info("File Successfully Renamed to : "+renamedFile);
-        }
-        else {
-            logger.info("Operation Failed to rename file : "+tokenRangeFile);
-        }
-
-        List<Partition> partitions = new ArrayList<Partition>();
-        BufferedReader reader = getfileReader(renamedFile);
-        String line = null;
-        while ((line = reader.readLine()) != null) {
-            if (line.startsWith("#")) {
-                continue;
-            }
-            String[] minMax = line.split(",");
-            try {
-                partitions.addAll(getSubPartitions(splitSize, new BigInteger(minMax[0]), new BigInteger(minMax[1]), 100));
-            } catch (Exception e) {
-                logger.error("Skipping partition: {}", line, e);
-            }
-        }
-
-        return partitions;
-    }
-
 }

src/main/java/com/datastax/cdm/properties/KnownProperties.java

@@ -141,10 +141,10 @@ public enum PropertyType {
     // Error handling
     //==========================================================================
     public static final String TOKEN_RANGE_EXCEPTION_DIR = "spark.tokenRange.exceptionDir"; // aaa/bbb
-    public static final String TOKEN_RANGE_EXCEPTION_FILE = "spark.input.partitionFile"; // aaa/bbb/filename
+    public static final String PARTITIONS_TOKEN_RANGE_FILE = "spark.input.partitionFile"; // aaa/bbb/filename
     static {
             types.put(TOKEN_RANGE_EXCEPTION_DIR, PropertyType.STRING);
-            types.put(TOKEN_RANGE_EXCEPTION_FILE, PropertyType.STRING);
+            types.put(PARTITIONS_TOKEN_RANGE_FILE, PropertyType.STRING);
     }
     //==========================================================================
     // Guardrails and Transformations

src/main/scala/com/datastax/cdm/job/BaseJob.scala

@@ -32,7 +32,6 @@ abstract class BaseJob[T: ClassTag] extends App {
   var maxPartition: BigInteger = _
   var coveragePercent: Int = _
   var numSplits: Int = _
-  var tokenRangeFile: String = _
 
   var parts: util.Collection[T] = _
   var slices: RDD[T] = _
@@ -70,18 +69,16 @@ abstract class BaseJob[T: ClassTag] extends App {
     maxPartition = getMaxPartition(propertyHelper.getString(KnownProperties.PARTITION_MAX), hasRandomPartitioner)
     coveragePercent = propertyHelper.getInteger(KnownProperties.TOKEN_COVERAGE_PERCENT)
     numSplits = propertyHelper.getInteger(KnownProperties.PERF_NUM_PARTS)
-    tokenRangeFile = propertyHelper.getString(KnownProperties.TOKEN_RANGE_EXCEPTION_FILE)
+    this.fileName = propertyHelper.getString(KnownProperties.PARTITIONS_TOKEN_RANGE_FILE)
     abstractLogger.info("PARAM -- Min Partition: " + minPartition)
     abstractLogger.info("PARAM -- Max Partition: " + maxPartition)
     abstractLogger.info("PARAM -- Number of Splits : " + numSplits)
     abstractLogger.info("PARAM -- Coverage Percent: " + coveragePercent)
-    abstractLogger.info("PARAM -- Token Range File: " + tokenRangeFile)
-
+    abstractLogger.info("PARAM -- Partition File: " + fileName)
     this.parts = getParts(numSplits)
     this.slices = sContext.parallelize(parts.asScala.toSeq, parts.size);
     abstractLogger.info("PARAM Calculated -- Total Partitions: " + parts.size())
     abstractLogger.info("Spark parallelize created : " + slices.count() + " slices!");
-
   }
 
   def getParts(pieces: Int): util.Collection[T]

src/main/scala/com/datastax/cdm/job/DiffPartitionsFromFile.scala

@@ -1,9 +1,14 @@
 package com.datastax.cdm.job
 
 object DiffPartitionsFromFile extends BasePartitionJob {
-  setup("Diff Partitions from File Job", new DiffJobSessionFactory(), tokenRangeFile)
-  execute()
-  finish()
+  setup("Diff Partitions from File Job", new DiffJobSessionFactory())
+  if ("".equals(this.fileName)) {
+    abstractLogger.error("Please set conf for spark.input.partitionFile ")
+  }
+  else {
+    execute()
+    finish()
+  }
 
   override def execute(): Unit = {
     slices.foreach(slice => {