Fix option casing issue, don't use DAX for DescribeTable

- Option keys are case sensitive have been lowercased by the time they
get to TableConnector
- DAX isn't capable of doing DescribeTale, so we ask for a non-DAX
client for that purpose

Author: rehevkor5

README.md

@@ -92,7 +92,9 @@ The following parameters can be set as options on the Spark reader object before
 
 - `readPartitions` number of partitions to split the initial RDD when loading the data into Spark. Defaults to the size of the DynamoDB table divided into chunks of `maxPartitionBytes`
 - `maxPartitionBytes` the maximum size of a single input partition. Default 128 MB
-- `defaultParallelism` the number of input partitions that can be read from DynamoDB simultaneously. Defaults to `sparkContext.defaultParallelism`
+- `defaultParallelism` the number of input partitions that can be read from or written to DynamoDB simultaneously.
+Read/write throughput will be limited by dividing it by this number. Set this to the number of CPU cores in your
+Spark job. Defaults to the value of `SparkContext#defaultParallelism`.
 - `targetCapacity` fraction of provisioned read capacity on the table (or index) to consume for reading, enforced by
 a rate limiter. Default 1 (i.e. 100% capacity).
 - `stronglyConsistentReads` whether or not to use strongly consistent reads. Default false.

src/main/scala/com/audienceproject/spark/dynamodb/connector/DynamoConnector.scala

@@ -31,24 +31,30 @@ import com.amazonaws.services.dynamodbv2.{AmazonDynamoDB, AmazonDynamoDBAsync, A
 import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClientBuilder
 import com.amazonaws.services.securitytoken.model.AssumeRoleRequest
 import org.apache.spark.sql.sources.Filter
+import org.slf4j.LoggerFactory
 
 private[dynamodb] trait DynamoConnector {
+    private val logger = LoggerFactory.getLogger(this.getClass)
 
     @transient private lazy val properties = sys.props
 
-    def getDynamoDB(region: Option[String] = None, roleArn: Option[String] = None, providerClassName: Option[String] = None): DynamoDB = {
-        val client: AmazonDynamoDB = getDynamoDBClient(region, roleArn, providerClassName)
+    def getDynamoDB(region: Option[String] = None, roleArn: Option[String] = None,
+                    providerClassName: Option[String] = None, omitDax: Boolean = false): DynamoDB = {
+        val client: AmazonDynamoDB = getDynamoDBClient(region, roleArn, providerClassName, omitDax)
         new DynamoDB(client)
     }
 
     private def getDynamoDBClient(region: Option[String] = None,
                                   roleArn: Option[String] = None,
-                                  providerClassName: Option[String]): AmazonDynamoDB = {
+                                  providerClassName: Option[String],
+                                  omitDax: Boolean = false): AmazonDynamoDB = {
         val chosenRegion = region.getOrElse(properties.getOrElse("aws.dynamodb.region", "us-east-1"))
         val credentials = getCredentials(chosenRegion, roleArn, providerClassName)
 
-        if (daxEndpoint.isEmpty) {
+        if (omitDax || daxEndpoint.isEmpty) {
+            logger.info("NOT using DAX")
             properties.get("aws.dynamodb.endpoint").map(endpoint => {
+                logger.debug(s"Using DynamoDB endpoint ${endpoint}")
                 AmazonDynamoDBClientBuilder.standard()
                     .withCredentials(credentials)
                     .withEndpointConfiguration(new EndpointConfiguration(endpoint, chosenRegion))
@@ -60,22 +66,26 @@ private[dynamodb] trait DynamoConnector {
                     .build()
             )
         } else {
+            logger.debug(s"Using DAX endpoint ${daxEndpoint}")
             AmazonDaxClientBuilder.standard()
                 .withEndpointConfiguration(daxEndpoint)
                 .withCredentials(credentials)
+                .withRegion(chosenRegion)
                 .build()
         }
 
     }
 
     def getDynamoDBAsyncClient(region: Option[String] = None,
                                roleArn: Option[String] = None,
-                               providerClassName: Option[String] = None): AmazonDynamoDBAsync = {
+                               providerClassName: Option[String] = None,
+                               omitDax: Boolean = false): AmazonDynamoDBAsync = {
         val chosenRegion = region.getOrElse(properties.getOrElse("aws.dynamodb.region", "us-east-1"))
         val credentials = getCredentials(chosenRegion, roleArn, providerClassName)
 
-        if (daxEndpoint.isEmpty) {
+        if (omitDax || daxEndpoint.isEmpty) {
             properties.get("aws.dynamodb.endpoint").map(endpoint => {
+                logger.debug(s"Using DynamoDB endpoint ${endpoint}")
                 AmazonDynamoDBAsyncClientBuilder.standard()
                     .withCredentials(credentials)
                     .withEndpointConfiguration(new EndpointConfiguration(endpoint, chosenRegion))
@@ -87,9 +97,11 @@ private[dynamodb] trait DynamoConnector {
                     .build()
             )
         } else {
+            logger.debug(s"Using DAX endpoint ${daxEndpoint}")
             AmazonDaxAsyncClientBuilder.standard()
                 .withEndpointConfiguration(daxEndpoint)
                 .withCredentials(credentials)
+                .withRegion(chosenRegion)
                 .build()
         }
     }

src/main/scala/com/audienceproject/spark/dynamodb/connector/TableConnector.scala

@@ -32,6 +32,12 @@ import org.apache.spark.sql.sources.Filter
 import scala.annotation.tailrec
 import scala.collection.JavaConverters._
 
+/**
+ *
+ * @param tableName
+ * @param parallelism
+ * @param parameters case sensitive Map, all keys have been lowercased
+ */
 private[dynamodb] class TableConnector(tableName: String, parallelism: Int, parameters: Map[String, String])
     extends DynamoConnector with DynamoWritable with Serializable {
 
@@ -43,10 +49,10 @@ private[dynamodb] class TableConnector(tableName: String, parallelism: Int, para
 
     override val filterPushdownEnabled: Boolean = filterPushdown
 
-    override val daxEndpoint: String = parameters.getOrElse("daxEndpoint", "").trim
+    override val daxEndpoint: String = parameters.getOrElse("daxendpoint", "").trim
 
     override val (keySchema, readLimit, writeLimit, itemLimit, totalSegments) = {
-        val table = getDynamoDB(region, roleArn, providerClassName).getTable(tableName)
+        val table = getDynamoDB(region, roleArn, providerClassName, omitDax = true).getTable(tableName)
         val desc = table.describe()
 
         // Key schema.

src/main/scala/com/audienceproject/spark/dynamodb/connector/TableIndexConnector.scala

@@ -39,7 +39,7 @@ private[dynamodb] class TableIndexConnector(tableName: String, indexName: String
 
     override val filterPushdownEnabled: Boolean = filterPushdown
 
-    override val daxEndpoint: String = parameters.getOrElse("daxEndpoint", "").trim
+    override val daxEndpoint: String = parameters.getOrElse("daxendpoint", "").trim
 
     override val (keySchema, readLimit, itemLimit, totalSegments) = {
         val table = getDynamoDB(region, roleArn, providerClassName).getTable(tableName)