viết một RDD vào HDFS trong một bối cảnh spark-streaming

Question

Tôi có một tia lửa chảy môi trường với tia lửa 1.2.0 nơi tôi lấy dữ liệu từ một thư mục địa phương và mỗi khi tôi tìm thấy một tập tin mới được thêm vào thư mục Tôi thực hiện một số chuyển đổi.

val ssc = new StreamingContext(sc, Seconds(10)) 
val data = ssc.textFileStream(directory) 

Để thực hiện phân tích của tôi trên dữ liệu DStream tôi phải biến nó thành một mảng

var arr = new ArrayBuffer[String](); 
    data.foreachRDD { 
    arr ++= _.collect() 
} 

Sau đó, tôi sử dụng dữ liệu thu được để trích xuất thông tin mà tôi muốn và lưu chúng trên HDFS.

val myRDD = sc.parallelize(arr) 
myRDD.saveAsTextFile("hdfs directory....") 

Kể từ khi tôi thực sự cần phải thao tác dữ liệu với một mảng nó không thể lưu dữ liệu trên HDFS với DStream.saveAsTextFiles("...") (mà sẽ làm việc tốt) và tôi phải lưu RDD nhưng với preocedure này cuối cùng tôi đã có tập tin đầu ra có sản phẩm nào được đặt tên bán 00000 vv ...

với một arr.foreach(println) tôi có thể xem kết quả chính xác của các transofmations.

nghi ngờ của tôi là tia lửa mà cố gắng ở mỗi đợt ghi dữ liệu trong cùng một file, xóa những gì đã được viết trước đó. Tôi đã cố gắng lưu trong một thư mục có tên động như myRDD.saveAsTextFile("folder" + System.currentTimeMillis().toString()) nhưng luôn luôn chỉ có một nếp gấp được tạo và các tệp đầu ra vẫn trống.

Làm thế nào tôi có thể viết một RDD vào HDFS trong một bối cảnh spark-streaming?

Answer 1

Bạn đang sử dụng Spark streaming trong một cách thức mà nó không được thiết kế. Tôi khuyên bạn nên sử dụng Spark để sử dụng cho trường hợp sử dụng của bạn hoặc điều chỉnh mã của bạn để nó hoạt động theo cách Spark. Thu thập mảng cho người lái xe đánh bại mục đích sử dụng một công cụ phân tán và làm cho ứng dụng của bạn có hiệu quả một máy (hai máy cũng sẽ gây ra nhiều chi phí hơn là chỉ xử lý dữ liệu trên một máy).

Tất cả những gì bạn có thể làm với một mảng, bạn có thể làm với Spark. Vì vậy, chỉ cần chạy tính toán của bạn bên trong dòng, phân phối trên các công nhân, và viết đầu ra của bạn bằng cách sử dụng DStream.saveAsTextFiles(). Bạn có thể sử dụng foreachRDD + saveAsParquet(path, overwrite = true) để ghi vào một tệp.

Answer 2

@vzamboni: Spark 1.5 + dataframes api có tính năng này:

dataframe.write().mode(SaveMode.Append).format(FILE_FORMAT).partitionBy("parameter1", "parameter2").save(path);