Angeben der Laufzeitabhängigkeit in Apache Spark

Question

Ich versuche, eine einfache Spark-App auszuführen, die Apache Tika verwendet, um Metadaten zu extrahieren. Ich habe ein schattiertes Glas mit Maven erstellt und die App läuft gut, aber die Tika-Parser werden zur Laufzeit nicht benutzt. Der App-Code ist unter:

SparkConf sparkConf = new SparkConf(); 
JavaSparkContext sc = new JavaSparkContext(sparkConf); 
if (args.length == 0 || args[0] == null) { 
    return; 
} 
JavaPairRDD<String,PortableDataStream> files = sc.binaryFiles(args[0]); 
List<Map<String,String>> results = files.map(new Function<Tuple2<String,PortableDataStream>, Map<String,String>>() { 

    public Map<String,String> call(Tuple2<String, PortableDataStream> stringPortableDataStreamTuple2) throws Exception { 
     DataInputStream des = stringPortableDataStreamTuple2._2.open(); 
     Tika tika = new Tika(); 
     Parser parser = new AutoDetectParser(); 
     BodyContentHandler handler = new BodyContentHandler(-1); 
     String detected = tika.detect(des); 
     Map<String,String> metadata = new HashMap(); 
     metadata.put("Type", detected); 
     Metadata tikaMetadata = new Metadata(); 
     parser.parse(des, handler, tikaMetadata, new ParseContext()); 
     String[] names = tikaMetadata.names(); 
     for (String name : names) { 
      metadata.put(name, tikaMetadata.get(name)); 
     } 
     return metadata; 
    } 
}).collect(); 
for (Map<String,String> o : results) { 
    for (String key : o.keySet()) { 
     System.out.println(key + " : " + o.get(key)); 
    } 
    System.out.println(); 
} 
sc.stop(); 

Der Ausgang ist immer so etwas wie:

Type : image/jpeg 
X-Parsed-By : org.apache.tika.parser.EmptyParser 
Content-Type : application/octet-stream 

Welche Tika-Parser gibt nicht verwendet werden. Nach der Dokumentation kann ich Runtime-Abhängigkeiten mit der spark.driver.extraClassPath Eigenschaft konfigurieren, so geändert ich meine $ SPARK_HOME/conf/Funken defaults.conf.template Datei zu haben:

spark.driver.extraClassPath /path/to/tika/jar/tika-app-1.13.jar 

Es ist noch nicht Arbeiten, kann mir jemand helfen, warum zu verstehen? Gibt es alternativ eine bessere Möglichkeit, Laufzeitabhängigkeiten anzugeben? Ich verwende die App derzeit nur über Spark-Submit im lokalen Modus und verwende Spark 1.6.1.

Answer 1

Sie könnten Abhängigkeit Gläser mit --jars hinzufügen, während antrag,

zum Beispiel

./bin/spark-submit \ 
    --class org.apache.spark.examples.SparkPi \ 
    --master yarn \ 
    --deploy-mode cluster \ 
    --executor-memory 20G \ 
    --num-executors 50 \ 
    --jars /path/to/tika/jar/tika-app-1.13.jar \ 
    /path/to/examples.jar 

Ways of adding jars for execution

Answer 2

Der Code, der in der map Funktion eines RDD ist ausgeführt auf ein Executor, nicht der Fahrer. Da Sie Tika im map verwenden, müssen Sie die notwendigen Gläser zu je Testamentsvollstrecker kopiert haben

Sie tun könnten, dass durch:

• Bevorzugte Lösung, wie in der Antwort von #WoodChopper mentiioned durch mit --jars

• oder wenn Sie es vorziehen, das Hinzufügen zu spak-defaults.conf die folgende Zeile: spark.executor.extraClassPath /path/to/tika/jar/tika-app-1.13.jar