incidente Spark durante la lettura di file JSON quando collegato con AWS-java-sdk

Question

Sia config.json essere un piccolo file JSON:

{ 
    "toto": 1 
} 

Ho fatto un semplice codice che legge il file JSON con sc.textFile (perché il file può essere S3, locale o HDFS, in modo da file di testo è conveniente)

import org.apache.spark.{SparkContext, SparkConf} 

object testAwsSdk { 
    def main(args:Array[String]):Unit = { 
    val sparkConf = new SparkConf().setAppName("test-aws-sdk").setMaster("local[*]") 
    val sc = new SparkContext(sparkConf) 
    val json = sc.textFile("config.json") 
    println(json.collect().mkString("\n")) 
    } 
} 

Il file SBT tirare solo spark-core biblioteca

libraryDependencies ++= Seq(
    "org.apache.spark" %% "spark-core" % "1.5.1" % "compile" 
) 

il programma funziona come previsto, scrivendo il contenuto di config.json sullo standard output.

Ora voglio collegare anche con aws-java-sdk, sdk di Amazon per accedere a S3.

libraryDependencies ++= Seq(
    "com.amazonaws" % "aws-java-sdk" % "1.10.30" % "compile", 
    "org.apache.spark" %% "spark-core" % "1.5.1" % "compile" 
) 

Eseguendo lo stesso codice, spark genera la seguente eccezione.

Exception in thread "main" com.fasterxml.jackson.databind.JsonMappingException: Could not find creator property with name 'id' (in class org.apache.spark.rdd.RDDOperationScope) 
at [Source: {"id":"0","name":"textFile"}; line: 1, column: 1] 
    at com.fasterxml.jackson.databind.JsonMappingException.from(JsonMappingException.java:148) 
    at com.fasterxml.jackson.databind.DeserializationContext.mappingException(DeserializationContext.java:843) 
    at com.fasterxml.jackson.databind.deser.BeanDeserializerFactory.addBeanProps(BeanDeserializerFactory.java:533) 
    at com.fasterxml.jackson.databind.deser.BeanDeserializerFactory.buildBeanDeserializer(BeanDeserializerFactory.java:220) 
    at com.fasterxml.jackson.databind.deser.BeanDeserializerFactory.createBeanDeserializer(BeanDeserializerFactory.java:143) 
    at com.fasterxml.jackson.databind.deser.DeserializerCache._createDeserializer2(DeserializerCache.java:409) 
    at com.fasterxml.jackson.databind.deser.DeserializerCache._createDeserializer(DeserializerCache.java:358) 
    at com.fasterxml.jackson.databind.deser.DeserializerCache._createAndCache2(DeserializerCache.java:265) 
    at com.fasterxml.jackson.databind.deser.DeserializerCache._createAndCacheValueDeserializer(DeserializerCache.java:245) 
    at com.fasterxml.jackson.databind.deser.DeserializerCache.findValueDeserializer(DeserializerCache.java:143) 
    at com.fasterxml.jackson.databind.DeserializationContext.findRootValueDeserializer(DeserializationContext.java:439) 
    at com.fasterxml.jackson.databind.ObjectMapper._findRootDeserializer(ObjectMapper.java:3666) 
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:3558) 
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2578) 
    at org.apache.spark.rdd.RDDOperationScope$.fromJson(RDDOperationScope.scala:82) 
    at org.apache.spark.rdd.RDDOperationScope$$anonfun$5.apply(RDDOperationScope.scala:133) 
    at org.apache.spark.rdd.RDDOperationScope$$anonfun$5.apply(RDDOperationScope.scala:133) 
    at scala.Option.map(Option.scala:145) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:133) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108) 
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:709) 
    at org.apache.spark.SparkContext.hadoopFile(SparkContext.scala:1012) 
    at org.apache.spark.SparkContext$$anonfun$textFile$1.apply(SparkContext.scala:827) 
    at org.apache.spark.SparkContext$$anonfun$textFile$1.apply(SparkContext.scala:825) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108) 
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:709) 
    at org.apache.spark.SparkContext.textFile(SparkContext.scala:825) 
    at testAwsSdk$.main(testAwsSdk.scala:11) 
    at testAwsSdk.main(testAwsSdk.scala) 
    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 com.intellij.rt.execution.application.AppMain.main(AppMain.java:140) 

Leggendo la pila, sembra che quando AWS-java-sdk è legata, sc.textFile rileva che il file è un file JSON e cercare di analizzarlo con Jackson assumendo un certo formato, che non riesce a trovare, naturalmente, . Ho bisogno di collegarmi con aws-java-sdk, quindi le mie domande sono:

1- Perché aggiungere aws-java-sdk modifica il comportamento di spark-core?

2- C'è una soluzione (il file può essere su HDFS, S3 o locale)?

Answer 1

Ha contattato l'assistenza Amazon. È un problema di depency con la biblioteca di Jackson. In SBT, ignorare Jackson:

libraryDependencies ++= Seq( 
"com.amazonaws" % "aws-java-sdk" % "1.10.30" % "compile", 
"org.apache.spark" %% "spark-core" % "1.5.1" % "compile" 
) 

dependencyOverrides ++= Set( 
"com.fasterxml.jackson.core" % "jackson-databind" % "2.4.4" 
) 

la loro risposta: Lo abbiamo fatto su un Mac, EC2 (RedHat AMI) e in secondo EMR (Amazon Linux). 3 ambienti diversi. La causa principale del problema è che sbt crea un grafico delle dipendenze e poi affronta il problema dei conflitti di versione rimuovendo la versione precedente e selezionando l'ultima versione della libreria dipendente. In questo caso, la scintilla dipende dalla versione 2.4 della libreria di jackson mentre l'SDS di AWS ha bisogno di 2.5. Quindi esiste un conflitto di versione e la versione di dipendenza di spark (che è più vecchia) di SRS sfrutta la versione di SDK di AWS (che è l'ultima).

Answer 2

Aggiungendo a Boris' answer, se non si desidera utilizzare una versione fissa di Jackson (magari in futuro verrà aggiornato Spark) ma ancora voglia di scartare l'uno da AWS, è possibile effettuare le seguenti operazioni:

libraryDependencies ++= Seq( 
    "com.amazonaws" % "aws-java-sdk" % "1.10.30" % "compile" excludeAll (
    ExclusionRule("com.fasterxml.jackson.core", "jackson-databind") 
), 
    "org.apache.spark" %% "spark-core" % "1.5.1" % "compile" 
)