Added config to separately add port & improved docs

Author: pravinbhat

pom.xml

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

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

@@ -12,13 +12,13 @@ class AbstractJob extends BaseJob {
   abstractLogger.info("PARAM -- Origin SSL Enabled: {}", sourceSSLEnabled);
   abstractLogger.info("PARAM -- Target SSL Enabled: {}", destinationSSLEnabled);
 
-  var sourceConnection = getConnection(true, sourceScbPath, sourceHost, sourceUsername, sourcePassword, sourceSSLEnabled,
+  var sourceConnection = getConnection(true, sourceScbPath, sourceHost, sourcePort, sourceUsername, sourcePassword, sourceSSLEnabled,
     sourceTrustStorePath, sourceTrustStorePassword, sourceTrustStoreType, sourceKeyStorePath, sourceKeyStorePassword, sourceEnabledAlgorithms);
 
-  var destinationConnection = getConnection(false, destinationScbPath, destinationHost, destinationUsername, destinationPassword, destinationSSLEnabled,
+  var destinationConnection = getConnection(false, destinationScbPath, destinationHost, destinationPort, destinationUsername, destinationPassword, destinationSSLEnabled,
     destinationTrustStorePath, destinationTrustStorePassword, destinationTrustStoreType, destinationKeyStorePath, destinationKeyStorePassword, destinationEnabledAlgorithms);
 
-  private def getConnection(isSource: Boolean, scbPath: String, host: String, username: String, password: String,
+  private def getConnection(isSource: Boolean, scbPath: String, host: String, port: String, username: String, password: String,
                             sslEnabled: String, trustStorePath: String, trustStorePassword: String, trustStoreType: String,
                             keyStorePath: String, keyStorePassword: String, enabledAlgorithms: String): CassandraConnector = {
     var connType: String = "Source"
@@ -36,7 +36,7 @@ class AbstractJob extends BaseJob {
         .set("spark.cassandra.input.consistency.level", consistencyLevel)
         .set("spark.cassandra.connection.config.cloud.path", scbPath))
     } else if (trustStorePath.nonEmpty) {
-      abstractLogger.info(connType + ": Connecting (with clientAuth) to Cassandra (or DSE) host: " + host);
+      abstractLogger.info(connType + ": Connecting (with clientAuth) to Cassandra (or DSE) host:port " + host + ":" + port);
 
       // Use defaults when not provided
       var enabledAlgorithmsVar = enabledAlgorithms
@@ -49,6 +49,7 @@ class AbstractJob extends BaseJob {
         .set("spark.cassandra.auth.password", password)
         .set("spark.cassandra.input.consistency.level", consistencyLevel)
         .set("spark.cassandra.connection.host", host)
+        .set("spark.cassandra.connection.port", port)
         .set("spark.cassandra.connection.ssl.enabled", "true")
         .set("spark.cassandra.connection.ssl.enabledAlgorithms", enabledAlgorithmsVar)
         .set("spark.cassandra.connection.ssl.trustStore.password", trustStorePassword)
@@ -59,13 +60,14 @@ class AbstractJob extends BaseJob {
         .set("spark.cassandra.connection.ssl.clientAuth.enabled", "true")
       )
     } else {
-      abstractLogger.info(connType + ": Connecting to Cassandra (or DSE) host: " + host);
+      abstractLogger.info(connType + ": Connecting to Cassandra (or DSE) host:port " + host + ":" + port);
 
       return CassandraConnector(config.set("spark.cassandra.auth.username", username)
         .set("spark.cassandra.connection.ssl.enabled", sslEnabled)
         .set("spark.cassandra.auth.password", password)
         .set("spark.cassandra.input.consistency.level", consistencyLevel)
-        .set("spark.cassandra.connection.host", host))
+        .set("spark.cassandra.connection.host", host)
+        .set("spark.cassandra.connection.port", port))
     }
 
   }

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

@@ -22,6 +22,7 @@ class BaseJob extends App {
 
   val sourceScbPath = Util.getSparkPropOrEmpty(sc, "spark.origin.scb")
   val sourceHost = Util.getSparkPropOrEmpty(sc, "spark.origin.host")
+  val sourcePort = Util.getSparkPropOr(sc, "spark.origin.port", "9042")
   val sourceUsername = Util.getSparkPropOrEmpty(sc, "spark.origin.username")
   val sourcePassword = Util.getSparkPropOrEmpty(sc, "spark.origin.password")
   val sourceSSLEnabled = Util.getSparkPropOr(sc, "spark.origin.ssl.enabled", "false")
@@ -34,6 +35,7 @@ class BaseJob extends App {
 
   val destinationScbPath = Util.getSparkPropOrEmpty(sc, "spark.target.scb")
   val destinationHost = Util.getSparkPropOrEmpty(sc, "spark.target.host")
+  val destinationPort = Util.getSparkPropOr(sc, "spark.target.port", "9042")
   val destinationUsername = Util.getSparkProp(sc, "spark.target.username")
   val destinationPassword = Util.getSparkProp(sc, "spark.target.password")
   val destinationSSLEnabled = Util.getSparkPropOr(sc, "spark.target.ssl.enabled", "false")

src/resources/sparkConf.properties

@@ -1,10 +1,14 @@
-# Origin cluster credentials
+# Origin cluster credentials (use "host + port" OR "secure-connect-bundle" but not both)
 spark.origin.host                                 localhost
+spark.origin.port                                 9042
+#spark.origin.scb                                 file:///aaa/bbb/secure-connect-enterprise.zip
 spark.origin.username                             some-username
 spark.origin.password                             some-secret-password
 spark.origin.keyspaceTable                        test.a1
 
-# Target cluster credentials
+# Target cluster credentials (use "host + port" OR "secure-connect-bundle" but not both)
+#spark.target.host                                localhost
+#spark.target.port                                9042
 spark.target.scb                                  file:///aaa/bbb/secure-connect-enterprise.zip
 spark.target.username                             client-id
 spark.target.password                             client-secret
@@ -29,12 +33,13 @@ spark.splitSize                                   10000
 spark.batchSize                                   10
 
 # Below 'query' properties are set based on table schema
-spark.query.origin                                partition-key,clustering-key,order-date,amount
-spark.query.origin.partitionKey                   partition-key
-spark.query.target.id                             partition-key,clustering-key
+spark.query.origin                                comma-separated-partition-key,comma-separated-clustering-key,comma-separated-other-columns
+spark.query.origin.partitionKey                   comma-separated-partition-key
+spark.query.target.id                             comma-separated-partition-key,comma-separated-clustering-key
+# Comma separated numeric data-type mapping (e.g. 'text' will map to '0') for all columns listed in "spark.query.origin"
 spark.query.types                                 9,1,4,3
 #############################################################################################################
-# Following are the supported data types and their corresponding [Cassandra data-types]
+# Following are the supported data types and their corresponding [Cassandra data-types] mapping
 #  0: ascii, text, varchar
 #  1: int
 #  2: bigint, counter