Tại sao foreach không mang lại bất cứ điều gì cho chương trình điều khiển?

Question

tôi đã viết chương trình này trong spark shell

val array = sc.parallelize(List(1, 2, 3, 4)) 
array.foreach(x => println(x)) 

này in một số báo cáo debug nhưng không phải là con số thực tế.

Đoạn code dưới đây hoạt động tốt

for(num <- array.take(4)) { 
    println(num) 
} 

Tôi hiểu rằng take là một hành động và do đó sẽ gây ra tia lửa để kích hoạt các tính lười biếng.

Nhưng foreach nên đã làm việc cùng một cách ... tại sao foreach không mang lại bất cứ điều gì trở lại từ tia lửa và bắt đầu thực hiện việc xử lý thực tế (thoát khỏi chế độ lười biếng)

Làm thế nào tôi có thể làm cho foreach trên RDD công việc?

Answer 1

Phương thức RDD.foreach trong Spark chạy trên cụm sao cho mỗi nhân viên chứa các hồ sơ này đang chạy các hoạt động trong foreach. I E. mã của bạn đang chạy, nhưng chúng đang in ra trên công nghệ Spark stdout, không phải trong trình điều khiển/phiên trình bao của bạn. Nếu bạn nhìn vào đầu ra (stdout) cho công nhân Spark của mình, bạn sẽ thấy những bản in này được in trên bảng điều khiển.

Bạn có thể xem thông tin chi tiết về người lao động bằng cách truy cập web gui đang chạy cho từng người thực thi đang chạy. Một URL ví dụ là http://workerIp:workerPort/logPage/?appId=app-20150303023103-0043&executorId=1&logType=stdout

Trong ví dụ Spark này lựa chọn để đặt tất cả các hồ sơ của RDD trong cùng một phân vùng.

Điều này có ý nghĩa nếu bạn nghĩ về nó - nhìn vào chữ ký chức năng cho foreach - nó không trả về bất cứ điều gì.

/** 
* Applies a function f to all elements of this RDD. 
*/ 
def foreach(f: T => Unit): Unit 

Đây thực sự là mục đích của foreach trong scala - được sử dụng để tác dụng phụ.

Khi bạn thu thập hồ sơ, bạn mang chúng trở lại trình điều khiển để thu thập/thực hiện thao tác chỉ chạy trên bộ sưu tập Scala trong trình điều khiển Spark - bạn có thể thấy đầu ra nhật ký như trình điều khiển tia lửa/vỏ tia lửa là gì để stdout trong phiên của bạn. Một trường hợp sử dụng của foreach có thể không rõ ràng ngay lập tức, ví dụ - nếu cho mỗi bản ghi trong RDD bạn muốn thực hiện một số hành vi bên ngoài, như gọi REST api, bạn có thể làm điều này trong foreach, sau đó mỗi Spark nhân viên sẽ gửi một cuộc gọi đến máy chủ API với giá trị. Nếu foreach đã mang lại hồ sơ, bạn có thể dễ dàng thổi ra bộ nhớ trong quá trình lái xe/vỏ. Bằng cách này bạn tránh những vấn đề này và có thể thực hiện các tác dụng phụ trên tất cả các mục trong RDD trên cụm.

Nếu bạn muốn xem những gì trong RDD tôi sử dụng;

array.collect.foreach(println) 
//Instead of collect, use take(...) or takeSample(...) if the RDD is large 

Answer 2

Bạn có thể sử dụng RDD.toLocalIterator() để mang dữ liệu cho tài xế (một phân vùng RDD tại một thời điểm):

val array = sc.parallelize(List(1, 2, 3, 4)) 
for(rec <- array.toLocalIterator) { println(rec) } 

Xem thêm

• Spark: Best practice for retrieving big data from RDD to local machine
• this blog post về toLocalIterator