स्पार्क-स्ट्रीमिंग संदर्भ में एचडीएफएस में एक आरडीडी लिखें

Question

मेरे पास स्पार्क 1.2.0 के साथ एक स्पार्क स्ट्रीमिंग वातावरण है जहां मैं स्थानीय फ़ोल्डर से डेटा पुनर्प्राप्त करता हूं और हर बार जब मैं फ़ोल्डर में एक नई फ़ाइल जोड़ता हूं तो मैं कुछ परिवर्तन करता हूं।

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

आदेश DStream डेटा पर अपने विश्लेषण प्रदर्शन करने के लिए मैं इसे किसी सरणी

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

तो मैं जानकारी मैं चाहता हूँ को निकालने के लिए और HDFS पर उन्हें बचाने के लिए प्राप्त डेटा का उपयोग में रूपांतरित करने के लिए है।

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

के बाद से मैं वास्तव में किसी सरणी यह ​​(जो ठीक काम करेगा) DStream.saveAsTextFiles("...") साथ HDFS पर डेटा सहेजने देती असंभव है के साथ डेटा में हेरफेर करने की जरूरत है और मैं RDD को बचाने के लिए है, लेकिन इस preocedure साथ मैं अंत में नामित खाली उत्पादन फ़ाइलें भाग -00000 आदि ...

arr.foreach(println) के साथ मैं ट्रांसफ़ोमेशन के सही परिणाम देख पा रहा हूं।

मेरा संदेह यह है कि स्पार्क एक ही फाइल में डेटा लिखने के लिए हर बैच पर कोशिश करता है, जो पहले लिखा गया था उसे हटा देता है। मैंने myRDD.saveAsTextFile("folder" + System.currentTimeMillis().toString()) जैसे गतिशील नामित फ़ोल्डर में सहेजने की कोशिश की लेकिन हमेशा केवल एक गुना बनाया गया है और आउटपुट फाइलें अभी भी खाली हैं।

मैं स्पार्क-स्ट्रीमिंग संदर्भ में एचडीएफएस में आरडीडी कैसे लिख सकता हूं?

Answer 1

आप स्पार्क स्ट्रीमिंग का उपयोग ऐसे तरीके से कर रहे हैं जिस पर इसे डिज़ाइन नहीं किया गया था। मैं या तो आपके उपयोग के मामले के लिए स्पार्क का उपयोग करके ड्रॉप की सिफारिश करता हूं, या अपना कोड अनुकूलित करता हूं ताकि यह स्पार्क तरीके से काम कर सके। ड्राइवर को सरणी एकत्र करना एक वितरित इंजन का उपयोग करने के उद्देश्य को हरा देता है और आपके ऐप को प्रभावी ढंग से सिंगल-मशीन बनाता है (दो मशीनें एक मशीन पर डेटा को प्रोसेस करने की तुलना में अधिक ओवरहेड का कारण बनती हैं)।

सब कुछ आप एक सरणी के साथ कर सकते हैं, आप स्पार्क के साथ कर सकते हैं। तो बस स्ट्रीम के अंदर अपनी गणनाएं चलाएं, श्रमिकों पर वितरित करें, और DStream.saveAsTextFiles() का उपयोग करके अपना आउटपुट लिखें। आप एक फ़ाइल को लिखने के लिए foreachRDD + saveAsParquet(path, overwrite = true) का उपयोग कर सकते हैं।

Answer 2

@vzamboni:

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