set access credentials on the Configuration, so reads can succeed

Author: koeninger

src/main/scala/com/databricks/spark/redshift/Parameters.scala

@@ -170,9 +170,28 @@ private [redshift] object Parameters extends Logging {
      * use of standard system properties, environment variables, or IAM role configuration if available.
      */
     def credentialsString(configuration: Configuration) = {
+      val ((_, accessKeyId), (_, secretAccessKey)) = credentialsTuple(configuration)
 
+      s"aws_access_key_id=$accessKeyId;aws_secret_access_key=$secretAccessKey"
+    }
+
+    /**
+     * Looks up "aws_access_key_id" and "aws_secret_access_key" in the parameter map
+     * and ensures they are set on the Configuration. If no credentials have been provided,
+     * this function will instead try using the Hadoop Configuration fs.* settings for the provided tempDir
+     * scheme, and if that also fails, it finally tries AWS DefaultCredentialsProviderChain, which makes
+     * use of standard system properties, environment variables, or IAM role configuration if available.
+     */
+    def setCredentials(configuration: Configuration): Unit = {
+      val ((accessKeyIdProp, accessKeyId), (secretAccessKeyProp, secretAccessKey)) = credentialsTuple(configuration)
+
+      configuration.setIfUnset(accessKeyIdProp, accessKeyId)
+      configuration.setIfUnset(secretAccessKeyProp, secretAccessKey)
+    }
+
+    private def credentialsTuple(configuration: Configuration) = {
       val scheme = new URI(tempDir).getScheme
-      val hadoopConfPrefix = s"fs.$scheme}"
+      val hadoopConfPrefix = s"fs.$scheme"
 
       val (accessKeyId, secretAccessKey) =
         if(parameters.contains("aws_access_key_id")) {
@@ -192,7 +211,7 @@ private [redshift] object Parameters extends Logging {
           }
         }
 
-      s"aws_access_key_id=$accessKeyId;aws_secret_access_key=$secretAccessKey"
+      ((s"$hadoopConfPrefix.awsAccessKeyId", accessKeyId), (s"$hadoopConfPrefix.awsSecretAccessKey", secretAccessKey))
     }
   }
 }

src/main/scala/com/databricks/spark/redshift/RedshiftRelation.scala

@@ -79,6 +79,7 @@ case class RedshiftRelation(jdbcWrapper: JDBCWrapper, params: MergedParameters,
 
   protected def makeRdd(schema: StructType): RDD[Row] = {
     val sc = sqlContext.sparkContext
+    params.setCredentials(sc.hadoopConfiguration)
     val rdd = sc.newAPIHadoopFile(params.tempPath, classOf[RedshiftInputFormat],
       classOf[java.lang.Long], classOf[Array[String]], sc.hadoopConfiguration)
     rdd.values.map(Conversions.rowConverter(schema))