PySpark verteilte Verarbeitung auf einem YARN-Cluster

Question

Ich habe Spark auf einem Cloudera CDH5.3-Cluster ausgeführt, mit YARN als Ressourcenmanager. Ich entwickle Spark-Apps in Python (PySpark).

Ich kann Jobs einreichen und sie laufen erfolgreich, aber sie scheinen nie auf mehr als einer Maschine zu laufen (die lokale Maschine, von der ich schicke).

Ich habe eine Vielzahl von Optionen ausprobiert, wie zum Beispiel --deploy-mode zu cluster und --master zu Garn-client und yarn-cluster, aber es scheint nie auf mehr als einem Server zu laufen.

Ich kann es auf mehr als einem Kern laufen lassen, indem man etwas wie --master local [8] übergibt, aber das verteilt offensichtlich die Verarbeitung nicht über mehrere Knoten.

Ich habe ein sehr einfaches Python-Skript Verarbeitung von Daten von HDFS etwa so:

import simplejson as json 
from pyspark import SparkContext 
sc = SparkContext("", "Joe Counter") 

rrd = sc.textFile("hdfs:///tmp/twitter/json/data/") 

data = rrd.map(lambda line: json.loads(line)) 

joes = data.filter(lambda tweet: "Joe" in tweet.get("text","")) 

print joes.count() 

und ich bin ein vorlegen Befehl wie Laufen:

spark-submit atest.py --deploy-mode client --master yarn-client 

Was kann ich den Job läuft, um sicherzustellen, tun parallel über den Cluster?

Answer 1

Können Sie die Argumente für den Befehl austauschen? Funken einreichen --deploy-Modus-Client --master Garn-Client atest.py

Wenn Sie den Hilfetext für den Befehl siehe:

Funken einreichen

Usage: spark-submit [options] <app jar | python file> 

Answer 2

Es scheint, dass PySpark nicht im verteilten Modus mit Spark/YARN läuft - Sie müssen Stand-alone-Spark mit einem Spark-Master-Server verwenden. In diesem Fall lief mein PySpark-Skript sehr gut über den Cluster mit einem Python-Prozess pro Kern/Knoten.

Answer 3

Ich glaube @ MrChristine ist korrekt - die Optionsflags, die Sie angeben, werden an Ihr Python-Skript übergeben, nicht an funke-submit. Außerdem sollten Sie --executor-cores und --num-executors angeben, da sie standardmäßig auf einem einzelnen Kern ausgeführt werden und zwei Executoren verwenden.

Answer 4

Es ist nicht wahr, dass Python-Skript nicht im Cluster-Modus ausgeführt wird. Ich bin mir nicht sicher über frühere Versionen, aber dies wird in der funke 2.2-Version auf Hortonworks-Cluster ausgeführt.

Befehl: Funken einreichen --master Garn --num-Vollstrecker 10 --executor-Kern 1 --driver-Speicher 5g /pyspark-example.py

Python-Code:

from pyspark import SparkConf, SparkContext 
from pyspark.sql import SQLContext 

conf = (SparkConf() 
     .setMaster("yarn") 
     .setAppName("retrieve data")) 
sc = SparkContext(conf = conf) 

sqlContext = SQLContext(sc) 
parquetFile = sqlContext.read.parquet("/<hdfs-path>/*.parquet") 

parquetFile.createOrReplaceTempView("temp") 
df1 = sqlContext.sql("select * from temp limit 5") 
df1.show() 
df1.write.save('/<hdfs-path>/test.csv', format='csv', mode='append') 
sc.stop() 

Ausgabe: Es ist groß, so dass ich nicht einfügen. Aber es läuft perfekt.