🐉 🧘 ✋🏼 अजगर बनाम अपाचे स्पार्क के लिए स्काला - अप्रत्याशित परिणाम के साथ अपेक्षित बेंचमार्क 🛎️ 🎴 😺

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

प्रतियोगिता के भाग के रूप में, प्रतिभागी सोशल नेटवर्क के समाचार फीड को सॉर्ट करने की समस्या को हल करते हैं और सॉर्ट सूची के सेट के रूप में समाधान अपलोड करते हैं। प्राप्त समाधान की गुणवत्ता की जांच करने के लिए, पहले, भरी हुई सूचियों में से प्रत्येक के लिए, आरओसी एयूसी की गणना की जाती है, और फिर औसत मूल्य प्रदर्शित किया जाता है। कृपया ध्यान दें कि आपको एक सामान्य आरओसी एयूसी की गणना करने की आवश्यकता नहीं है, लेकिन प्रत्येक उपयोगकर्ता के लिए एक व्यक्तिगत - इस समस्या को हल करने के लिए कोई तैयार डिज़ाइन नहीं है, इसलिए आपको एक विशेष फ़ंक्शन लिखना होगा। व्यवहार में दो दृष्टिकोणों की तुलना करने का एक अच्छा कारण।

एक तुलना मंच के रूप में, हम स्थानीय मोड में लॉन्च किए गए चार कोर और स्पार्क के साथ एक क्लाउड कंटेनर का उपयोग करेंगे, और हम अपाचे ज़ेपेलिन के माध्यम से इसके साथ काम करेंगे। कार्यक्षमता की तुलना करने के लिए, हम PySpark और Scala Spark में समान कोड को मिरर करेंगे। [यहाँ] चलो डेटा लोड करके शुरू करते हैं।

data = sqlContext.read.csv("sna2019/modelCappedSubmit") trueData = sqlContext.read.csv("sna2019/collabGt") toValidate = data.withColumnRenamed("_c1", "submit") \ .join(trueData.withColumnRenamed("_c1", "real"), "_c0") \ .withColumnRenamed("_c0", "user") \ .repartition(4).cache() toValidate.count()

 val data = sqlContext.read.csv("sna2019/modelCappedSubmit") val trueData = sqlContext.read.csv("sna2019/collabGt") val toValidate = data.withColumnRenamed("_c1", "submit") .join(trueData.withColumnRenamed("_c1", "real"), "_c0") .withColumnRenamed("_c0", "user") .repartition(4).cache() toValidate.count()

मानक API का उपयोग करते समय, कोड की लगभग पूरी पहचान val कीवर्ड तक उल्लेखनीय है। ऑपरेटिंग समय काफी भिन्न नहीं है। अब हम जिस UDF की आवश्यकता है, उसे निर्धारित करने का प्रयास करते हैं।

 parse = sqlContext.udf.register("parse", lambda x: [int(s.strip()) for s in x[1:-1].split(",")], ArrayType(IntegerType())) def auc(submit, real): trueSet = set(real) scores = [1.0 / (i + 1) for i,x in enumerate(submit)] labels = [1.0 if x in trueSet else 0.0 for x in submit] return float(roc_auc_score(labels, scores)) auc_udf = sqlContext.udf.register("auc", auc, DoubleType())

 val parse = sqlContext.udf.register("parse", (x : String) => x.slice(1,x.size - 1).split(",").map(_.trim.toInt)) case class AucAccumulator(height: Int, area: Int, negatives: Int) val auc_udf = sqlContext.udf.register("auc", (byScore: Seq[Int], gt: Seq[Int]) => { val byLabel = gt.toSet val accumulator = byScore.foldLeft(AucAccumulator(0, 0, 0))((accumulated, current) => { if (byLabel.contains(current)) { accumulated.copy(height = accumulated.height + 1) } else { accumulated.copy(area = accumulated.area + accumulated.height, negatives = accumulated.negatives + 1) } }) (accumulator.area).toDouble / (accumulator.negatives * accumulator.height) })

किसी विशिष्ट फ़ंक्शन को लागू करते समय, यह स्पष्ट है कि पायथन अधिक संक्षिप्त है, मुख्य रूप से अंतर्निहित स्किकिट-लर्न फ़ंक्शन का उपयोग करने की क्षमता के कारण। हालांकि, अप्रिय क्षण हैं - आपको स्पष्ट रूप से वापसी मूल्य के प्रकार को निर्दिष्ट करना होगा, जबकि स्काला में यह स्वचालित रूप से निर्धारित होता है। चलो ऑपरेशन करते हैं:

 toValidate.select(auc_udf(parse("submit"), parse("real"))).groupBy().avg().show()

 toValidate.select(auc_udf(parse($"submit"), parse($"real"))).groupBy().avg().show()

कोड लगभग समान दिखता है, लेकिन परिणाम हतोत्साहित कर रहे हैं।

पाइस्पार्क पर क्रियान्वयन ने स्काला पर दो सेकंड के बजाय डेढ़ मिनट काम किया, यानी पायथन 45 गुना धीमा हो गया । दौड़ते समय, शीर्ष 4 सक्रिय पायथन प्रक्रियाएं दिखाता है जो पूरी गति से चल रही हैं, और इससे पता चलता है कि ग्लोबल इंटरप्रेटर लॉक को यहां कोई समस्या नहीं है। लेकिन! शायद समस्या स्किटिट-लर्न के आंतरिक कार्यान्वयन में है - चलो मानक पुस्तकालयों का सहारा लिए बिना, सचमुच पायथन कोड को पुन: पेश करने का प्रयास करें।

 def auc(submit, real): trueSet = set(real) height = 0 area = 0 negatives = 0 for candidate in submit: if candidate in trueSet: height = height + 1 else: area = area + height negatives = negatives + 1 return float(area) / (negatives * height) auc_udf_modified = sqlContext.udf.register("auc_modified", auc, DoubleType()) toValidate.select(auc_udf_modified(parse("submit"), parse("real"))).groupBy().avg().show()

प्रयोग दिलचस्प परिणाम दिखाता है। एक ओर, इस दृष्टिकोण के साथ, उत्पादकता को समतल किया गया था, लेकिन दूसरी तरफ, लैकोनिज़्म गायब हो गया। प्राप्त परिणामों से संकेत मिल सकता है कि जब अतिरिक्त C ++ मॉड्यूल का उपयोग करके पायथन में काम कर रहे हैं, तो संदर्भों के बीच स्विच करने के लिए महत्वपूर्ण ओवरहेड दिखाई देते हैं। बेशक, जावा / स्काला में जेएनआई का उपयोग करते समय समान ओवरहेड होता है, हालांकि, मुझे उपयोग करते समय 45 बार गिरावट के उदाहरणों से निपटना नहीं था।

अधिक विस्तृत विश्लेषण के लिए, हम दो अतिरिक्त प्रयोगों को अंजाम देंगे: पैकेज के आह्वान से योगदान को मापने के लिए स्पार्क के बिना शुद्ध पायथन का उपयोग करना, और ओवरहेड को परिशोधन करने के लिए स्पार्क में बढ़े हुए डेटा आकार के साथ और अधिक सटीक तुलना प्राप्त करना।

 def parse(x): return [int(s.strip()) for s in x[1:-1].split(",")] def auc(submit, real): trueSet = set(real) height = 0 area = 0 negatives = 0 for candidate in submit: if candidate in trueSet: height = height + 1 else: area = area + height negatives = negatives + 1 return float(area) / (negatives * height) def sklearn_auc(submit, real): trueSet = set(real) scores = [1.0 / (i + 1) for i,x in enumerate(submit)] labels = [1.0 if x in trueSet else 0.0 for x in submit] return float(roc_auc_score(labels, scores))

स्थानीय पायथन और पंडों के साथ प्रयोग ने अतिरिक्त पैकेजों का उपयोग करते समय महत्वपूर्ण ओवरहेड की धारणा की पुष्टि की - जब स्किट-लर्न का उपयोग किया जाता है, तो गति 20 गुना से अधिक घट जाती है। हालाँकि, 20 45 नहीं है - चलो डेटा को "भड़काने" की कोशिश करते हैं और स्पार्क प्रदर्शन की फिर से तुलना करते हैं।

 k4 = toValidate.union(toValidate) k8 = k4.union(k4) m1 = k8.union(k8) m2 = m1.union(m1) m4 = m2.union(m2).repartition(4).cache() m4.count()

नई तुलना पायथन पर स्कैला कार्यान्वयन की गति लाभ को 7-8 गुना - 7 सेकंड बनाम 55 से दर्शाती है। अंत में, चलो "सबसे तेजी से पायथन में है" का प्रयास करें - सरणी के योग की गणना करने के लिए संख्यात्मक :

 import numpy numpy_sum = sqlContext.udf.register("numpy_sum", lambda x: float(numpy.sum(x)), DoubleType())

 val my_sum = sqlContext.udf.register("my_sum", (x: Seq[Int]) => x.map(_.toDouble).sum)

फिर से एक महत्वपूर्ण मंदी - स्काला के 5 सेकंड बनाम पाइथन के 80 सेकंड। सारांशित करते हुए, हम निम्नलिखित निष्कर्ष निकाल सकते हैं:

जबकि PySpark मानक API के ढांचे के भीतर संचालित होता है, यह वास्तव में Scala की गति में तुलनीय हो सकता है।
जब विशिष्ट तर्क उपयोगकर्ता परिभाषित क्रियाओं के रूप में प्रकट होता है, तो PySpark का प्रदर्शन कम हो जाता है। पर्याप्त जानकारी के साथ, जब डेटा ब्लॉक के लिए प्रसंस्करण समय कई सेकंड से अधिक हो जाता है, तो पायथन की व्याख्या करने पर प्रक्रियाओं और अपशिष्ट संसाधनों के बीच डेटा को स्थानांतरित करने की आवश्यकता के कारण अजगर कार्यान्वयन 5-10 धीमा है।
यदि C ++ मॉड्यूल में लागू अतिरिक्त फ़ंक्शन का उपयोग दिखाई देता है, तो अतिरिक्त कॉल लागत उत्पन्न होती हैं, और पायथन और स्काला के बीच का अंतर 10-50 गुना तक बढ़ जाता है।

परिणामस्वरूप, अजगर के सभी आकर्षण के बावजूद, स्पार्क के साथ संयोजन में इसका उपयोग हमेशा उचित नहीं लगता है। यदि पायथन ओवरहेड को महत्वपूर्ण बनाने के लिए बहुत अधिक डेटा नहीं है, तो आपको इस बारे में सोचना चाहिए कि क्या यहां स्पार्क की आवश्यकता है? यदि बहुत अधिक डेटा है, लेकिन मानक स्पार्क एसक्यूएल एपीआई के ढांचे के भीतर प्रसंस्करण होता है, तो यहां पायथन की आवश्यकता है?

यदि बहुत अधिक डेटा है और अक्सर ऐसे कार्यों से निपटना पड़ता है जो SQL API की सीमाओं से परे जाते हैं, तो PySpark का उपयोग करते समय समान कार्य करने के लिए, आपको क्लस्टर को काफी बढ़ाना होगा। उदाहरण के लिए, Odnoklassniki के लिए, स्पार्क क्लस्टर के लिए पूंजीगत व्यय की लागत कई लाख रूबल से बढ़ेगी। और अगर आप पायथन इकोसिस्टम पुस्तकालयों की उन्नत क्षमताओं का लाभ उठाने की कोशिश करते हैं, यानी, धीमा होने का जोखिम केवल समय पर नहीं है, बल्कि परिमाण का एक क्रम है।

वेक्टर किए गए कार्यों की अपेक्षाकृत नई कार्यक्षमता का उपयोग करके कुछ त्वरण प्राप्त किया जा सकता है। इस मामले में, यूडीएफ इनपुट को एक पंक्ति नहीं खिलाया जाता है, लेकिन पंडित डेटाफ्रेम के रूप में कई पंक्तियों का एक पैकेट। हालांकि, इस कार्यक्षमता का विकास अभी तक पूरा नहीं हुआ है , और इस मामले में भी अंतर महत्वपूर्ण होगा।

डेटा इंजीनियरों की एक व्यापक टीम को बनाए रखने के लिए एक विकल्प होगा, अतिरिक्त कार्यों के साथ डेटा वैज्ञानिकों की जरूरतों को जल्दी से पूरा करने में सक्षम। या अपने आप को स्काला दुनिया में डुबो देना, क्योंकि यह इतना मुश्किल नहीं है: कई आवश्यक उपकरण पहले से मौजूद हैं , प्रशिक्षण कार्यक्रम दिखाई देते हैं जो कि PySpark से परे हैं।

अजगर बनाम अपाचे स्पार्क के लिए स्काला - अप्रत्याशित परिणाम के साथ अपेक्षित बेंचमार्क

More articles: