Elaborazione distribuita di PySpark su un cluster YARN

Question

Spark è in esecuzione su un cluster CDH5.3 Cloudera, utilizzando YARN come gestore risorse. Sto sviluppando app Spark in Python (PySpark).

Posso inoltrare lavori e funzionano con successo, tuttavia non sembrano mai funzionare su più di una macchina (la macchina locale inviata da).

Ho provato diverse opzioni, come impostare --diploy-mode in cluster e --master in filato-client e filato-cluster, ma non sembra mai funzionare su più di un server.

Posso farlo funzionare su più di un core passando qualcosa come --master local [8], ma ovviamente non distribuisce l'elaborazione su più nodi.

ho una molto semplicemente Python elaborazione dei dati di script da HDFS in questo modo:

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() 

E Sono in esecuzione di un comando di presentare come:

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

Che cosa posso fare per garantire le piste di lavoro in parallelo attraverso il cluster?

Answer 1

È possibile scambiare gli argomenti per il comando? scintilla presentare client --deploy-mode --master filo-client atest.py

Se si vede il testo di aiuto per il comando:

scintilla presentare

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

Answer 2

Sembra che PySpark non funzioni in modalità distribuita utilizzando Spark/YARN - è necessario utilizzare Spark standalone con un server Spark Master. In questo caso, il mio script PySpark ha funzionato molto bene attraverso il cluster con un processo Python per core/node.

Answer 3

Credo @ MrChristine è corretto: i flag di opzione specificati vengono passati al tuo script python, non a spark-submit. Inoltre, ti consigliamo di specificare --executor-cores e --num-executors poiché per impostazione predefinita verrà eseguito su un singolo core e utilizzerà due esecutori.

Answer 4

Non è vero che lo script Python non viene eseguito in modalità cluster. Non sono sicuro delle versioni precedenti, ma questo è in esecuzione nella versione 2.2 di spark sul cluster Hortonworks.

Comando: scintilla presentare filato --master --num-esecutori 10 --executor-core 1 --driver-memory 5g /pyspark-example.py

codice Python:

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() 

Uscita: È grande quindi non sto incollando. Ma funziona perfettamente.