Tại sao tôi không thể đọc từ ứng dụng AWS S3 trong ứng dụng Spark nữa?

Question

Tôi đã nâng cấp lên Apache Spark 1.5.1 nhưng tôi không chắc liệu điều đó có gây ra hay không. Tôi có các phím truy cập của tôi trong quá trình kích hoạt mà luôn hoạt động.

Exception in thread "main" java.lang.NoSuchMethodError: org.jets3t.service.impl.rest.httpclient.RestS3Service.<init>(Lorg/jets3t/service/security/AWSCredentials;)V 

    SQLContext sqlContext = new SQLContext(sc); 
    DataFrame df = sqlContext.read() 
     .format("com.databricks.spark.csv") 
     .option("inferSchema", "true") 
     .load("s3n://ossem-replication/gdelt_data/event_data/" + args[0]); 

    df.write() 
     .format("com.databricks.spark.csv") 
     .save("/user/spark/ossem_data/gdelt/" + args[0]); 

Thêm lỗi bên dưới. Có một lớp không chứa phương thức để có nghĩa là các phụ thuộc không khớp. Có vẻ như jets3t không chứa phương thức RestS3Service (Lorg/jets3t/service/security/AWSCredentials;) V Có thể giải thích điều này cho tôi không?

Exception in thread "main" java.lang.NoSuchMethodError: org.jets3t.service.impl.rest.httpclient.RestS3Service.<init>(Lorg/jets3t/service/security/AWSCredentials;)V 
    at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.initialize(Jets3tNativeFileSystemStore.java:60) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:497) 
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187) 
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102) 
    at org.apache.hadoop.fs.s3native.$Proxy24.initialize(Unknown Source) 
    at org.apache.hadoop.fs.s3native.NativeS3FileSystem.initialize(NativeS3FileSystem.java:272) 
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2596) 
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91) 
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2630) 
    at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2612) 
    at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:370) 
    at org.apache.hadoop.fs.Path.getFileSystem(Path.java:296) 
    at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:256) 
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:228) 
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313) 
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:207) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 
    at scala.Option.getOrElse(Option.scala:120) 
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 
    at scala.Option.getOrElse(Option.scala:120) 
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 
    at org.apache.spark.rdd.RDD$$anonfun$take$1.apply(RDD.scala:1277) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108) 
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:306) 
    at org.apache.spark.rdd.RDD.take(RDD.scala:1272) 
    at org.apache.spark.rdd.RDD$$anonfun$first$1.apply(RDD.scala:1312) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108) 
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:306) 
    at org.apache.spark.rdd.RDD.first(RDD.scala:1311) 
    at com.databricks.spark.csv.CsvRelation.firstLine$lzycompute(CsvRelation.scala:101) 
    at com.databricks.spark.csv.CsvRelation.firstLine(CsvRelation.scala:99) 
    at com.databricks.spark.csv.CsvRelation.inferSchema(CsvRelation.scala:82) 
    at com.databricks.spark.csv.CsvRelation.<init>(CsvRelation.scala:42) 
    at com.databricks.spark.csv.DefaultSource.createRelation(DefaultSource.scala:74) 
    at com.databricks.spark.csv.DefaultSource.createRelation(DefaultSource.scala:39) 
    at com.databricks.spark.csv.DefaultSource.createRelation(DefaultSource.scala:27) 
    at org.apache.spark.sql.execution.datasources.ResolvedDataSource$.apply(ResolvedDataSource.scala:125) 
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:114) 
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:104) 
    at com.bah.ossem.spark.GdeltSpark.main(GdeltSpark.java:20) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:497) 

Answer 1

Tôi đã gặp vấn đề tương tự nhưng với Spark 1.6 và tôi đang sử dụng Scala thay vì Java. Lý do của lỗi này là Spark Core có phiên bản ứng dụng khách Hadoop 2.2 và cài đặt cụm Spark mà tôi đã sử dụng là 1.6. Tôi phải thực hiện các thay đổi sau để làm cho nó hoạt động.

1. Thay đổi phụ thuộc hadoop khách hàng để 2.6 (Phiên bản của Hadoop Tôi đã sử dụng)

   "org.apache.hadoop" % "hadoop-client" % "2.6.0", 
   

2. Bao gồm thư viện hadoop-AWS trong lọ mỡ Spark của tôi như là sự phụ thuộc này không còn nằm trong thư viện Hadoop trong 1.6

   "org.apache.hadoop" % "hadoop-aws" % "2.6.0", 
   

3. Xuất khóa AWS và bí mật làm biến môi trường.

4. Xác định cấu hình Hadoop sau trong SparkConf

   val sparkContext = new SparkContext(sparkConf) 
   val hadoopConf = sparkContext.hadoopConfiguration 
   hadoopConf.set("fs.s3.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem") 
   hadoopConf.set("fs.s3.awsAccessKeyId", sys.env.getOrElse("AWS_ACCESS_KEY_ID", "")) 
   hadoopConf.set("fs.s3.awsSecretAccessKey", sys.env.getOrElse("AWS_SECRET_ACCESS_KEY", ""))