问必须将AWS访问密钥ID和秘密访问密钥指定为用户名或密码(火花、Parquet、S3)

EN

package com.knx.rtb.sample

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions._

// One method for defining the schema of an RDD is to make a case class with the desired column
// names and types.
case class Record(key: Int, value: String)

object SparkSql {
  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setAppName("SparkSql")
    val sc = new SparkContext(sparkConf)
    sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", "accesskey")
    sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", "secretKey+JJbat7uEQtX/")

    val sqlContext = new SQLContext(sc)

    // Importing the SQL context gives access to all the SQL functions and implicit conversions.
    import sqlContext.implicits._

    val df = sc.parallelize((1 to 100).map(i => Record(i, s"val_$i"))).toDF()

    //if I remove this line; then I got the error
    df.write.parquet("s3n://adx-test/hdfs/pair.parquet")

    // Read in parquet file.  Parquet files are self-describing so the schmema is preserved.
    val parquetFile = sqlContext.read.parquet("s3n://adx-test/hdfs/pair.parquet")

    // Queries can be run using the DSL on parequet files just like the original RDD.
    parquetFile.where($"key" === 1).select($"value".as("a")).collect().foreach(println)

    // These files can also be registered as tables.
    parquetFile.registerTempTable("parquetFile")
    println("Result of Parquet file:")
    sqlContext.sql("SELECT * FROM parquetFile").collect().foreach(println)

    sc.stop()
  }
}