यार्न पर अपाचे स्पार्क: बड़ी मात्रा में इनपुट डेटा फ़ाइलों (स्पार्क में एकाधिक इनपुट फाइलों को गठबंधन करें)

Question

कार्यान्वयन के लिए एक सर्वोत्तम अभ्यास की आवश्यकता है। ऑपरेटिंग वातावरण इस प्रकार है:

• लॉग डेटा फ़ाइल अनियमित आता है।
• लॉग डेटा फ़ाइल का आकार 3.9 केबी से 8.5 एमबी तक है। औसत लगभग 1 एमबी है।
• डेटा फ़ाइल के रिकॉर्ड की संख्या 13 लाइनों से 22000 लाइनों तक है। औसत 2700 लाइनें है।
• डेटा फ़ाइल को एकत्रीकरण से पहले पोस्ट-प्रोसेस किया जाना चाहिए।
• पोस्ट-प्रोसेसिंग एल्गोरिदम बदला जा सकता है।
• पोस्ट-प्रोसेस की गई फ़ाइल को मूल डेटा फ़ाइल के साथ अलग से प्रबंधित किया जाता है, क्योंकि पोस्ट-प्रोसेसिंग एल्गोरिदम बदला जा सकता है।
• दैनिक एकत्रीकरण किया जाता है। सभी पोस्ट-प्रोसेस की गई डेटा फ़ाइल को रिकॉर्ड-दर-रिकॉर्ड और एकत्रीकरण (औसत, अधिकतम न्यूनतम ...) फ़िल्टर किया जाना चाहिए।
• चूंकि एकत्रीकरण ठीक है, इसलिए एकत्रीकरण के बाद रिकॉर्ड्स की संख्या इतनी छोटी नहीं है। यह मूल रिकॉर्ड की संख्या का आधा हिस्सा हो सकता है।
• एक बिंदु पर, के बाद संसाधित फ़ाइल की संख्या 200,000 के बारे में हो सकता है।
• एक डेटा फ़ाइल को व्यक्तिगत रूप से हटाया जा सकता है।

एक परीक्षण में, मैं 160,000 के बाद संसाधित ग्लोब पथ के साथ() sc.textFile के साथ शुरू स्पार्क द्वारा फ़ाइलों को प्रोसेस करने की कोशिश की, यह चालक की प्रक्रिया पर OutOfMemory अपवाद के साथ विफल रहा है।

इस तरह के डेटा को संभालने का सबसे अच्छा अभ्यास क्या है? क्या मुझे पोस्ट-प्रोसेस किए गए डेटा को सहेजने के लिए सादे फाइलों के बजाय एचबीएएस का उपयोग करना चाहिए?

Answer 1

हमने अपना लोडर लिखा है। यह HDFS में छोटे फाइलों के साथ हमारी समस्या को हल किया। यह हैडोप CombineFileInputFormat का उपयोग करता है। हमारे मामले में यह 100000 लगभग 3000 से मानचित्रकारों की संख्या कम और काफी तेजी काम कर दिया।

https://github.com/RetailRocket/SparkMultiTool

उदाहरण:

import ru.retailrocket.spark.multitool.Loaders 
val sessions = Loaders.combineTextFile(sc, "file:///test/*") 
// or val sessions = Loaders.combineTextFile(sc, conf.weblogs(), size = 256, delim = "\n") 
// where size is split size in Megabytes, delim - line break character 
println(sessions.count()) 

Answer 2

मैं यकीन है कि कारण आपके हो रही OOM क्योंकि बहुत से छोटे फ़ाइलों से निपटने में है हूँ। क्या आप चाहते हैं इनपुट फ़ाइलें गठबंधन करने के लिए ताकि आप इतने सारे विभाजन नहीं मिलता है। मैं अपनी नौकरियों को लगभग 10k विभाजन तक सीमित करने की कोशिश करता हूं।

textFile के बाद, आप .coalesce(10000, false) उपयोग कर सकते हैं ...% नहीं 100 सुनिश्चित करें कि हालांकि काम करेंगे क्योंकि यह कुछ समय के बाद से मैं इसे किया है हो गया है, तो कृपया मुझे बताएं। तो

sc.textFile(path).coalesce(10000, false) 

Answer 3

आप उपयोग कर सकते हैं इस

सबसे पहले आप कर सकते हैं एक बफर/S3 पथ/की सूची HDFS या स्थानीय पथ के लिए एक ही मिलता है

आप तो अमेज़ॅन एस 3 के साथ कोशिश कर रहे हैं:

import scala.collection.JavaConverters._ 
import java.util.ArrayList 
import com.amazonaws.services.s3.AmazonS3Client 
import com.amazonaws.services.s3.model.ObjectListing 
import com.amazonaws.services.s3.model.S3ObjectSummary 
import com.amazonaws.services.s3.model.ListObjectsRequest 

def listFiles(s3_bucket:String, base_prefix : String) = { 
    var files = new ArrayList[String] 

    //S3 Client and List Object Request 
    var s3Client = new AmazonS3Client(); 
    var objectListing: ObjectListing = null; 
    var listObjectsRequest = new ListObjectsRequest(); 

    //Your S3 Bucket 
    listObjectsRequest.setBucketName(s3_bucket) 

    //Your Folder path or Prefix 
    listObjectsRequest.setPrefix(base_prefix) 

    //Adding s3:// to the paths and adding to a list 
    do { 
     objectListing = s3Client.listObjects(listObjectsRequest); 
     for (objectSummary <- objectListing.getObjectSummaries().asScala) { 
     files.add("s3://" + s3_bucket + "/" + objectSummary.getKey()); 
     } 
     listObjectsRequest.setMarker(objectListing.getNextMarker()); 
    } while (objectListing.isTruncated()); 

    //Removing Base Directory Name 
    files.remove(0) 

    //Creating a Scala List for same 
    files.asScala 
    } 

अब कोड का निम्न भाग को यह सूची वस्तु दर्रा, ध्यान दें: अनुसूचित जाति SQLContext

var df: DataFrame = null; 
    for (file <- files) { 
    val fileDf= sc.textFile(file) 
    if (df!= null) { 
     df= df.unionAll(fileDf) 
    } else { 
     df= fileDf 
    } 
    } 

अब की एक वस्तु आपके पास एक अंतिम एकीकृत RDD यानी df

वैकल्पिक है, और आप भी कर सकते हैं एक भी BigRDD

val files = sc.textFile(filename, 1).repartition(1) 

में यह पुनर्विभाजन पुनर्विभाजन हमेशा काम करता है: डी