PySpark được xử lý phân tán trên cụm YARN

Question

Tôi có Spark chạy trên cụm Cloudera CDH5.3, sử dụng YARN làm trình quản lý tài nguyên. Tôi đang phát triển ứng dụng Spark trong Python (PySpark).

Tôi có thể gửi công việc và chạy thành công, tuy nhiên chúng dường như không bao giờ chạy trên nhiều máy (máy cục bộ mà tôi gửi).

Tôi đã thử nhiều tùy chọn, như cài đặt --chế độ-chế độ thành cụm và - quản trị đối với khách hàng sợi và sợi, nhưng dường như nó chưa bao giờ chạy trên nhiều máy chủ.

Tôi có thể làm cho nó chạy trên nhiều lõi bằng cách truyền một cái gì đó như --master local [8], nhưng rõ ràng là không phân phối quá trình xử lý qua nhiều nút.

Tôi có một rất đơn giản xử lý dữ liệu kịch bản Python từ HDFS như vậy:

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

Và tôi đang chạy một lệnh nộp như:

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

tôi có thể làm gì để đảm bảo chạy việc song song trên cụm sao?

Answer 1

Bạn có thể trao đổi các đối số cho lệnh không? spark-nộp --deploy-mode client --master sợi-client atest.py

Nếu bạn thấy văn bản trợ giúp cho lệnh:

spark-nộp

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

Answer 2

Dường như PySpark không chạy ở chế độ phân tán bằng Spark/YARN - bạn cần sử dụng Spark độc lập với máy chủ Spark Master. Trong trường hợp đó, kịch bản PySpark của tôi chạy rất tốt trên toàn bộ cluster với một tiến trình Python trên mỗi lõi/nút.

Answer 3

Tôi tin rằng @ MrChristine là đúng - các cờ tùy chọn mà bạn chỉ định đang được chuyển tới tập lệnh python của bạn, không phải để kích hoạt. Ngoài ra, bạn sẽ muốn chỉ định --executor-cores và --num-executors vì theo mặc định nó sẽ chạy trên một lõi đơn và sử dụng hai trình thực thi.

Answer 4

Không đúng là tập lệnh python không chạy ở chế độ cụm. Tôi không chắc chắn về các phiên bản trước đó nhưng điều này đang thực hiện trong phiên bản tia lửa 2.2 trên cụm Hortonworks.

Command: spark-nộp sợi --master --num-Chấp hành viên 10 --executor-lõi 1 --driver bộ nhớ 5g /pyspark-example.py

Python Mã số:

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

Đầu ra: Lớn nên tôi không dán. Nhưng nó chạy hoàn hảo.