Merge pull request apache#24 from sven0726/2.2-xspark

spark2.2.0 支持ak,sk 注入

Author: OopsOutOfMemory

core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala

@@ -84,21 +84,11 @@ class SparkHadoopUtil extends Logging {
     // the behavior of the old implementation of this code, for backwards compatibility.
     if (conf != null) {
       // Explicitly check for S3 environment variables
-      val keyId = System.getenv("AWS_ACCESS_KEY_ID")
-      val accessKey = System.getenv("AWS_SECRET_ACCESS_KEY")
-      if (keyId != null && accessKey != null) {
-        hadoopConf.set("fs.s3.awsAccessKeyId", keyId)
-        hadoopConf.set("fs.s3n.awsAccessKeyId", keyId)
-        hadoopConf.set("fs.s3a.access.key", keyId)
-        hadoopConf.set("fs.s3.awsSecretAccessKey", accessKey)
-        hadoopConf.set("fs.s3n.awsSecretAccessKey", accessKey)
-        hadoopConf.set("fs.s3a.secret.key", accessKey)
-
-        val sessionToken = System.getenv("AWS_SESSION_TOKEN")
-        if (sessionToken != null) {
-          hadoopConf.set("fs.s3a.session.token", sessionToken)
-        }
-      }
+      initWithAWSConf(hadoopConf)
+
+      // Check for Qiniu AK SK
+      initWithQiniuConf(hadoopConf)
+
       // Copy any "spark.hadoop.foo=bar" system properties into conf as "foo=bar"
       conf.getAll.foreach { case (key, value) =>
         if (key.startsWith("spark.hadoop.")) {
@@ -110,6 +100,33 @@ class SparkHadoopUtil extends Logging {
     }
   }
 
+  private def initWithAWSConf(hadoopConf: Configuration) = {
+    if (System.getenv("AWS_ACCESS_KEY_ID") != null &&
+      System.getenv("AWS_SECRET_ACCESS_KEY") != null) {
+      val keyId = System.getenv("AWS_ACCESS_KEY_ID")
+      val accessKey = System.getenv("AWS_SECRET_ACCESS_KEY")
+
+      hadoopConf.set("fs.s3.awsAccessKeyId", keyId)
+      hadoopConf.set("fs.s3n.awsAccessKeyId", keyId)
+      hadoopConf.set("fs.s3a.access.key", keyId)
+      hadoopConf.set("fs.s3.awsSecretAccessKey", accessKey)
+      hadoopConf.set("fs.s3n.awsSecretAccessKey", accessKey)
+      hadoopConf.set("fs.s3a.secret.key", accessKey)
+    }
+  }
+
+  private def initWithQiniuConf(hadoopConf: Configuration) = {
+    if (System.getenv("QINIU_ACCESS_KEY") != null &&
+      System.getenv("QINIU_SECRET_KEY") != null) {
+      val accessKey = System.getenv("QINIU_ACCESS_KEY")
+      val secretKey = System.getenv("QINIU_SECRET_KEY")
+
+      hadoopConf.set("fs.qiniu.access.key", accessKey)
+      hadoopConf.set("fs.qiniu.secret.key", secretKey)
+      logInfo("Success init conf with qiniu ak sk.")
+    }
+  }
+
   /**
    * Return an appropriate (subclass) of Configuration. Creating config can initializes some Hadoop
    * subsystems.