Làm thế nào để sử dụng cả Scala và Python trong cùng một dự án Spark?

Question

Có thể để ống Spark RDD tới Python?

Vì tôi cần thư viện python để thực hiện một số phép tính trên dữ liệu của mình, nhưng dự án Spark chính của tôi dựa trên Scala. Có cách nào kết hợp cả hai hoặc để cho python truy cập cùng một ngữ cảnh tia lửa?

Answer 1

Bạn thực sự có thể đường ống ra một kịch bản python sử dụng Scala và Spark và một kịch bản Python thường xuyên.

test.py

#!/usr/bin/python 

import sys 

for line in sys.stdin: 
    print "hello " + line 

spark-shell (scala)

val data = List("john","paul","george","ringo") 

val dataRDD = sc.makeRDD(data) 

val scriptPath = "./test.py" 

val pipeRDD = dataRDD.pipe(scriptPath) 

pipeRDD.foreach(println) 

Output

chào john

chào Ringo

chào george

chào paul

Answer 2

Nếu tôi hiểu chính xác, miễn là bạn lấy dữ liệu từ scala và chuyển nó thành RDD hoặc SparkContext thì bạn sẽ có thể sử dụng pyspark để thao tác dữ liệu bằng API Spark Python.

Ngoài ra còn có một programming guide mà bạn có thể làm theo để sử dụng các ngôn ngữ khác nhau trong vòng spark

Answer 3

Bạn có thể chạy mã Python qua ống trong Spark. Với pipe(), bạn có thể viết một phép biến đổi RDD đọc từng phần tử RDD từ đầu vào tiêu chuẩn như String, thao tác String theo lệnh script và sau đó ghi kết quả là String vào đầu ra tiêu chuẩn.

SparkContext.addFile (đường dẫn), chúng tôi có thể thêm danh sách tệp cho mỗi nút công nhân để tải xuống khi bắt đầu công việc Spark.Tất cả nút công nhân sẽ có bản sao của tập lệnh, do đó chúng tôi sẽ nhận được thao tác song song bằng đường ống. Chúng ta cần phải cài đặt tất cả các thư viện và sự phụ thuộc trước khi nó trên tất cả các nút công nhân và người thực hiện.

Ví dụ:

Python file: Mã để làm dữ liệu đầu vào thành chữ hoa

#!/usr/bin/python 
import sys 
for line in sys.stdin: 
    print line.upper() 

Spark Mã: Đối với Piping dữ liệu

val conf = new SparkConf().setAppName("Pipe") 
val sc = new SparkContext(conf) 
val distScript = "/path/on/driver/PipeScript.py" 
val distScriptName = "PipeScript.py" 
sc.addFile(distScript) 
val ipData = sc.parallelize(List("asd","xyz","zxcz","sdfsfd","Ssdfd","Sdfsf")) 
val opData = ipData.pipe(SparkFiles.get(distScriptName)) 
opData.foreach(println)