Bir kıvılcım akışı içeriğinde HDFS'ye bir RDD yazın

Question

Yerel bir klasörden veri aldığım ve her dönüşümü gerçekleştirdiğim klasöre yeni bir dosya bulduğum kıvılcım 1.2.0 ile kıvılcım akışı ortamım var. DStream verilere benim analizi gerçekleştirmek amacıyla

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

Ben

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

Sonra istediğim bilgiyi ayıklamak ve HDF'ler üzerine onları kurtarmak için elde edilen verileri kullanmak bir Array haline dönüştürmek zorundayız.

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

Gerçekten (ince çalışacak olan) DStream.saveAsTextFiles("...") ile HDF'ler verileri kaydetmek imkansız bir Array ile verileri işlemek için gerektiğinden ve RDD kurtarmak zorunda ama bu preocedure ile nihayet adında boş çıktı dosyaları var bölüm-00000 vb ...

Bir arr.foreach(println) ile Transofmasyonların doğru sonuçlarını görebiliyorum.

Şüphelim, bu kıvılcım, daha önce yazılanları silerek, aynı dosyada veri yazmak için her parti üzerinde çalışır. myRDD.saveAsTextFile("folder" + System.currentTimeMillis().toString()) gibi bir dinamik adlandırılmış klasöre kaydetmeye çalıştım, ancak her zaman yalnızca bir kat oluşturulur ve çıktı dosyaları hala boş.

Bir kıvılcım akışı bağlamında HDFS'ye nasıl bir RDD yazabilirim?

Answer 1

Spark Streaming'i, tasarlanmadığı bir şekilde kullanıyorsunuz. Kullanım durumunuz için Spark'i kullanarak ya da kodunuzu uyarlamanızı öneririz. Diziyi sürücüye toplamak, dağıtılmış bir motoru kullanmanın amacını ortadan kaldırır ve uygulamanızı tek makineli olarak etkin hale getirir (iki makine aynı zamanda verileri tek bir makinede işlemekten çok daha fazla çalışmaya neden olur).

Bir dizi ile yapabileceğiniz her şey Spark ile yapabilirsiniz. Bu yüzden, iş akışlarınızı, işyerinde dağıtılan akış içinde çalıştırın ve çıktınızı DStream.saveAsTextFiles() kullanarak yazın. Tek bir dosyaya yazmak için foreachRDD + saveAsParquet(path, overwrite = true)'u kullanabilirsiniz.

Answer 2

@vzamboni:

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