Спарк Сохранение результатов в HDFS

Question

Я столкнулся странное поведение при сохранении моего разобранный файл XML, используя искру с одним ведущим и трех рабочих узлов HDFS, проблема

когда я разобран XMLFILE и пытается сохранить в HDFS, то файл не может сохранить все результаты анализа.

и когда я казнен же код с локальным Режиме, указав

sc = SparkContext("local", "parser") 

and the spark-submit will be ./bin/spark-submit xml_parser.py 

этого запуск дает 117mb разобран файлу на HDFS с полными записями.

И в случае выполнения кода в режиме искрового-клиент, то я сделал следующее,

sc = SparkContext("spark://master:7077", "parser") 

И искра представить это,

./bin/spark-submit --master yarn-client --deploy-mode client --driver-memory 7g --executor-memory 4g --executor-cores 2 xml_parser.py 1000 

дает мне файл 19mb на HDFS с неполными записями ,

для сохранения результата в обоих случаях я использую rdd.saveAsTextFile ("HDFS: //")

я использую spark1.6.1-hadoop2.6 и Apache Hadoop 2.7.2

может кто-нибудь мне помочь. Я не понимаю, почему это происходит. У меня есть следующие sparkCluster,

1-мастер 8GbRAM

2-workerNode1 8GbRAM

3-WorkerNode2 8GbRAM

4-workerNode3 8GbRAM

и я настроен выше кластера над Hadoop-2,7.2 с 1 мастером и 3 DataNode,

если JPS I На severNode дает мне

24097 Мастер

21652 Jps

23398 NameNode

23799 ResourceManager

23630 SecondaryNameNode

JPS На всех DataNodes,

8006 Worker

7819 NodeManager

27164 Jps

7678 DataNode

Проверяя мастер HadoopNameNode Ui: 9000 дают мне три живых DataNode,

, проверив SparkMaster Ui на хозяине: 7077 дает мне трех живых работников

пожалуйста смотрите,

sc = SpakContext("spark://master:7077", "parser") 
-------------------------------------------- 
contains the logic of XMLParsing 
-------------------------------------------- 
and I am appending the result in one list like, 
cc_list.append([final_cll, Date,Time,int(cont[i]), float(values[i]),0]) 
Now I am Parallelizing the above cc_list like 
parallel_list = sc.parallelize(cc_list) 
parallel_list.saveAsTextFile("hdfs://master:9000/ some path") 
Now I am Doing some operations here. 
new_list = sc.textFile("hdfs://localhost:9000/some path/part-00000).map(lambda line:line.split(',')) 

result = new_list.map(lambda x: (x[0]+', '+x[3],float(x[4]))).sortByKey('true').coalesce(1) 
result = result.map(lambda x:x[0]+','+str(x[1])) 
result = result.map(lambda x: x.lstrip('[').rstrip(']').replace(' ','')).saveAsTextFile("hdfs://master:9000/some path1)) 

Answer 1

К сожалению для такого глупого вопроса здесь. на самом деле я считаю, две проблемы

1) во время работы на нескольких работника,

parallel_list = sc.parallelize(cc_list) 

создает 4-5 часть файлов и parallel_list сохраняется в HDFS с частично 00000 расстаться-00004, и при загрузке parallel_list и можно видеть выше в коде

new_list = sc.textFile(pathto parallel_list/part-00000) ==> so it was taking only the first part. 

2) во время работы на localMode,

parallel_list = sc.parallelize(cc_list) was creating only one part file so i was able to pick whole file at one stroke. 

Так во время работы на искре с рабочими я придумываю два решений

1) Я просто добавил * неполный при создании new_list из parallel_list

2) увеличило spark.akka.frameSize до 10000 по прохождение --conf spark.akka.frameSize = 1000 с искрой представить.