Merge pull request #52 from abinaya21/fix-cluster-mode-issue

[ISSUE-51] Fix error in accessing config when running job in cluster mode

Author: pravinbhat

pom.xml

@@ -3,7 +3,7 @@
 
   <groupId>datastax.astra.migrate</groupId>
   <artifactId>cassandra-data-migrator</artifactId>
-  <version>3.0.1</version>
+  <version>3.0.2</version>
   <packaging>jar</packaging>
 
   <properties>

src/main/scala/datastax/astra/migrate/DiffData.scala

@@ -3,18 +3,19 @@ package datastax.astra.migrate
 import com.datastax.spark.connector.cql.CassandraConnector
 import org.slf4j.LoggerFactory
 
+import org.apache.spark.SparkConf
 import scala.collection.JavaConversions._
 
 object DiffData extends AbstractJob {
 
   val logger = LoggerFactory.getLogger(this.getClass.getName)
   logger.info("Started Data Validation App")
 
-  diffTable(sourceConnection, destinationConnection)
+  diffTable(sourceConnection, destinationConnection, sc)
 
   exitSpark
 
-  private def diffTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector) = {
+  private def diffTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector, config: SparkConf) = {
     val partitions = SplitPartitions.getRandomSubPartitions(splitSize, minPartition, maxPartition, Integer.parseInt(coveragePercent))
     logger.info("PARAM Calculated -- Total Partitions: " + partitions.size())
     val parts = sContext.parallelize(partitions.toSeq, partitions.size);
@@ -23,11 +24,11 @@ object DiffData extends AbstractJob {
     parts.foreach(part => {
       sourceConnection.withSessionDo(sourceSession =>
         destinationConnection.withSessionDo(destinationSession =>
-          DiffJobSession.getInstance(sourceSession, destinationSession, sc)
+          DiffJobSession.getInstance(sourceSession, destinationSession, config)
             .getDataAndDiff(part.getMin, part.getMax)))
     })
 
-    DiffJobSession.getInstance(null, null, sc).printCounts(true);
+    DiffJobSession.getInstance(null, null, config).printCounts(true);
   }
 
 }

src/main/scala/datastax/astra/migrate/Migrate.scala

@@ -2,6 +2,7 @@ package datastax.astra.migrate
 
 import com.datastax.spark.connector.cql.CassandraConnector
 import org.slf4j.LoggerFactory
+import org.apache.spark.SparkConf
 
 import scala.collection.JavaConversions._
 
@@ -12,11 +13,11 @@ object Migrate extends AbstractJob {
   val logger = LoggerFactory.getLogger(this.getClass.getName)
   logger.info("Started Migration App")
 
-  migrateTable(sourceConnection, destinationConnection)
+  migrateTable(sourceConnection, destinationConnection, sc)
 
   exitSpark
 
-  private def migrateTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector) = {
+  private def migrateTable(sourceConnection: CassandraConnector, destinationConnection: CassandraConnector, config: SparkConf) = {
     val partitions = SplitPartitions.getRandomSubPartitions(splitSize, minPartition, maxPartition, Integer.parseInt(coveragePercent))
     logger.info("PARAM Calculated -- Total Partitions: " + partitions.size())
     val parts = sContext.parallelize(partitions.toSeq, partitions.size);
@@ -25,7 +26,7 @@ object Migrate extends AbstractJob {
     parts.foreach(part => {
       sourceConnection.withSessionDo(sourceSession =>
         destinationConnection.withSessionDo(destinationSession =>
-          CopyJobSession.getInstance(sourceSession, destinationSession, sc)
+          CopyJobSession.getInstance(sourceSession, destinationSession, config)
             .getDataAndInsert(part.getMin, part.getMax)))
     })