डेटा साइंस के क्षेत्र में पारंपरिक उपकरण आर और पायथन जैसी भाषाएं हैं - आराम से वाक्य रचना और मशीन सीखने और डेटा प्रोसेसिंग के लिए बड़ी संख्या में पुस्तकालय आपको जल्दी से कुछ काम करने वाले समाधान प्राप्त करने की अनुमति देते हैं। हालांकि, ऐसी स्थितियां हैं जब इन उपकरणों की सीमाएं एक महत्वपूर्ण बाधा बन जाती हैं - सबसे पहले, अगर प्रसंस्करण गति और / या वास्तव में बड़े डेटा सेट के साथ काम करने में उच्च प्रदर्शन प्राप्त करना आवश्यक है। इस मामले में, विशेषज्ञ को अनिच्छा से "डार्क साइड" की मदद से मुड़ना होगा और "औद्योगिक" प्रोग्रामिंग भाषाओं में उपकरणों को कनेक्ट करना होगा: स्काला , जावा और सी ++ ।

लेकिन क्या यह पक्ष इतना अंधेरा है? विकास के वर्षों में, "औद्योगिक" डेटा विज्ञान के उपकरण ने एक लंबा सफर तय किया है और आज से 2-3 साल पहले वे अपने स्वयं के संस्करणों से काफी अलग हैं। आइए SNA हैकथॉन 2019 कार्य के उदाहरण का उपयोग करके यह जानने का प्रयास करें कि स्केल + स्पार्क इकोसिस्टम पायथन डेटा साइंस के अनुरूप हो सकता है।

एसएनए हैकथॉन 2019 के ढांचे के भीतर , प्रतिभागी तीन "विषयों" में से एक में सामाजिक नेटवर्क के उपयोगकर्ता के समाचार फ़ीड को छाँटने की समस्या को हल करते हैं: पाठ, चित्र या फीचर लॉग के डेटा का उपयोग करना। इस प्रकाशन में, हम देखेंगे कि स्पार्क में शास्त्रीय मशीन लर्निंग टूल्स का उपयोग करते हुए संकेतों के लॉग के आधार पर किसी समस्या को हल करना कैसे संभव है।

समस्या को हल करने में, हम किसी मॉडल का विकास करते समय किसी डेटा विश्लेषण विशेषज्ञ के मानक तरीके से गुजरेंगे:

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

अपनी "यात्रा" के दौरान, हम ज़ेपेलिन इंटरैक्टिव नोटबुक, स्पार्क एमएल मशीन लर्निंग लाइब्रेरी और उसके विस्तार PravdaML , ग्राफएक्स ग्राफिंग पैकेज, वेगास विज़ुअलाइज़ेशन लाइब्रेरी और निश्चित रूप से, अपाचे स्पार्क को इसके सभी गौरव के रूप में ऐसे उपकरणों से परिचित करेंगे। )। सभी कोड और प्रयोग के परिणाम Zepl सहयोगी नोटबुक प्लेटफॉर्म पर उपलब्ध हैं।

डेटा लोड हो रहा है

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

इसलिए, पहला कदम - दिन पर डेटा के वितरण की जांच करना - बॉक्सिंग टूल द्वारा आसानी से किया जाता है:

val train = sqlContext.read.parquet("/events/hackatons/SNAHackathon/2019/collabTrain") z.show(train.groupBy($"date").agg( functions.count($"instanceId_userId").as("count"), functions.countDistinct($"instanceId_userId").as("users"), functions.countDistinct($"instanceId_objectId").as("objects"), functions.countDistinct($"metadata_ownerId").as("owners")) .orderBy("date")) 

Zeppelin में संबंधित ग्राफ क्या प्रदर्शित करेगा :

मुझे कहना होगा कि स्काला सिंटैक्स काफी लचीला है, और समान कोड लग सकता है, उदाहरण के लिए, इस तरह:

 val train = sqlContext.read.parquet("/events/hackatons/SNAHackathon/2019/collabTrain") z.show( train groupBy $"date" agg( count($"instanceId_userId") as "count", countDistinct($"instanceId_userId") as "users", countDistinct($"instanceId_objectId") as "objects", countDistinct($"metadata_ownerId") as "owners") orderBy "date" ) 

यहां एक महत्वपूर्ण चेतावनी दी जानी चाहिए: जब एक बड़ी टीम में काम करते हैं, जहां हर कोई अपने स्वयं के स्वाद के दृष्टिकोण से विशेष रूप से स्काला-कोड लिखने के लिए संपर्क करता है, तो संचार अधिक कठिन होता है। इसलिए कोड शैली की एकीकृत अवधारणा विकसित करना बेहतर है।

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

 z.show(filteredTrain .groupBy($"instanceId_userId").count .groupBy("count").agg(functions.log(functions.count("count")).as("withCount")) .orderBy($"withCount".desc) .limit(100) .orderBy($"count")) 

बहुत लंबी पूंछ के साथ एक वितरण को घातीय के करीब देखने की उम्मीद है। ऐसे कार्यों में, एक नियम के रूप में, गतिविधि के विभिन्न स्तरों वाले उपयोगकर्ताओं के लिए मॉडल को विभाजित करके काम की गुणवत्ता में सुधार प्राप्त करना संभव है। यह जाँचने के लिए कि क्या यह करने योग्य है, परीक्षण सेट में उपयोगकर्ता द्वारा वस्तुओं की संख्या के वितरण की तुलना करें:

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

 //  ,     "",   , //     val testSimilar = sc.broadcast(filteredTrain.groupBy($"instanceId_userId") .agg( functions.count("feedback").as("count"), functions.sum(functions.expr("IF(array_contains(feedback, 'Liked'), 1.0, 0.0)")).as("sum") ) .where("count > sum AND sum > 0") .select("instanceId_userId").rdd.map(_.getInt(0)).collect.sorted) //           // User Defined Function val isTestSimilar = sqlContext.udf.register("isTestSimilar", (x: Int) => java.util.Arrays.binarySearch(testSimilar.value, x) >= 0) 

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

पहली एमएल पाइपलाइन

अगले चरण में, हम क्रियाओं और विशेषताओं पर मूल आँकड़ों की गणना करने का प्रयास करेंगे। लेकिन इसके लिए हमें स्पार्कएमएल की क्षमताओं की आवश्यकता है, इसलिए पहले हम इसकी सामान्य वास्तुकला को देखेंगे:

SparkML निम्नलिखित अवधारणाओं के आधार पर बनाया गया है:

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

एमएल-एल्गोरिदम के गठन के लिए इस तरह के एक दृष्टिकोण से एक स्पष्ट मॉड्यूलर संरचना और अच्छा प्रजनन क्षमता प्राप्त करने में मदद मिलती है - दोनों मॉडल और पाइपलाइनों को बचाया जा सकता है।

शुरू करने के लिए, हम एक सरल पाइपलाइन का निर्माण करेंगे जिसके साथ हम प्रशिक्षण सेट में उपयोगकर्ताओं के कार्यों (प्रतिक्रिया क्षेत्र) के वितरण के आंकड़ों की गणना करते हैं:

 val feedbackAggregator = new Pipeline().setStages(Array( //         (feedback)  one-hot  new MultinominalExtractor().setInputCol("feedback").setOutputCol("feedback"), //       new VectorStatCollector() .setGroupByColumns("date").setInputCol("feedback") .setPercentiles(Array(0.1,0.5,0.9)), //        new VectorExplode().setValueCol("feedback") )).fit(train) z.show(feedbackAggregator .transform(filteredTrain) .orderBy($"date", $"feedback")) 

इस पाइपलाइन में, PravdaML की कार्यक्षमता सक्रिय रूप से उपयोग की जाती है - स्पार्कएमएल के लिए विस्तारित उपयोगी ब्लॉकों के साथ पुस्तकालय, अर्थात्:

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

कार्य का परिणाम यह दर्शाता है कि डेटासेट में कक्षाएं संतुलित नहीं हैं, हालांकि, लक्ष्य के लिए असंतुलन पसंद किया गया वर्ग चरम नहीं है:

परीक्षण के समान उपयोगकर्ताओं के बीच समान वितरण का विश्लेषण (दोनों लॉग में "सकारात्मक" और "नकारात्मक" होने) से पता चलता है कि यह सकारात्मक वर्ग के पक्षपाती है:

संकेतों का सांख्यिकीय विश्लेषण

अगले चरण में, हम विशेषताओं के सांख्यिकीय गुणों का एक विस्तृत विश्लेषण करेंगे। इस समय हमें एक बड़े कन्वेयर की आवश्यकता है:

 val statsAggregator = new Pipeline().setStages(Array( new NullToDefaultReplacer(), //          new AutoAssembler() .setColumnsToExclude( (Seq("date", "feedback") ++ train.schema.fieldNames.filter(_.endsWith("Id")) : _*)) .setOutputCol("features"), new VectorStatCollector() .setGroupByColumns("date").setInputCol("features") .setPercentiles(Array(0.1,0.5,0.9)), new VectorExplode().setValueCol("features") )) 

चूंकि अब हमें एक अलग क्षेत्र के साथ काम करने की आवश्यकता नहीं है, लेकिन एक ही बार में सभी विशेषताओं के साथ, हम दो और उपयोगी प्रवीण उपयोगिताओं का उपयोग करेंगे:

• NullToDefaultReplacer आपको उनके डिफ़ॉल्ट मान (संख्या के लिए 0, तार्किक चर के लिए गलत, आदि) के साथ डेटा में लापता तत्वों को बदलने की अनुमति देता है। यदि आप यह रूपांतरण नहीं करते हैं, तो परिणामी वैक्टर में NaN मान दिखाई देगा, जो कई एल्गोरिदम के लिए घातक है (हालांकि, उदाहरण के लिए, XGBoost इस से बच सकता है)। शून्य के साथ प्रतिस्थापित करने का एक विकल्प औसत के साथ प्रतिस्थापित किया जा सकता है, यह NaNToMeanReplacerEstimator में लागू किया गया है।
• AutoAssembler एक बहुत शक्तिशाली उपयोगिता है जो तालिका लेआउट का विश्लेषण करती है और प्रत्येक स्तंभ के लिए एक वेक्टराइज़ेशन योजना का चयन करती है जो स्तंभ के प्रकार से मेल खाती है।

परिणामी पाइपलाइन का उपयोग करते हुए, हम तीन सेट (प्रशिक्षण, उपयोगकर्ता फ़िल्टर और परीक्षण के साथ प्रशिक्षण) और अलग-अलग फ़ाइलों में सहेजने के लिए आँकड़ों की गणना करते हैं:

 //   (   AutoAssembler  ) val trained = statsAggregator.fit(filteredTrain) //       - ,     . trained .transform(filteredTrain .withColumn("date", //  ,      ,     , //        All   functions.explode(functions.array(functions.lit("All"), $"date")))) .coalesce(7).write.mode("overwrite").parquet("sna2019/featuresStat") trained .transform(filteredTrain .where(isTestSimilar($"instanceId_userId")) .withColumn("date", functions.explode(functions.array(functions.lit("All"), $"date")))) .coalesce(7).write.mode("overwrite").parquet("sna2019/filteredFeaturesStat") trained .transform(filteredTest.withColumn("date", functions.explode(functions.array(functions.lit("All"), $"date")))) .coalesce(3).write.mode("overwrite").parquet("sna2019/testFeaturesStat") 

विशेषता आंकड़ों के साथ तीन डेटासेट प्राप्त करने के बाद, हम निम्नलिखित बातों का विश्लेषण करते हैं:

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

इन पहलुओं को स्पष्ट करने के लिए, इस तरह के अनुरोध से हमें मदद मिलेगी:

 def compareWithTest(data: DataFrame) : DataFrame = { data.where("date = 'All'") .select( $"features", //         // ( ) functions.log($"features_mean" / $"features_p50").as("skewenes"), //    90-      //    90-  —    functions.log( ($"features_max" - $"features_p90") / ($"features_p90" - $"features_p50")).as("outlieres"), //       ,  //    ($"features_nonZeros" / $"features_count").as("train_fill"), $"features_mean".as("train_mean")) .join(testStat.where("date = 'All'") .select($"features", $"features_mean".as("test_mean"), ($"features_nonZeros" / $"features_count").as("test_fill")), Seq("features")) //          .withColumn("meanDrift", (($"train_mean" - $"test_mean" ) / ($"train_mean" + $"test_mean"))) //      .withColumn("fillDrift", ($"train_fill" - $"test_fill") / ($"train_fill" + $"test_fill")) } //         val comparison = compareWithTest(trainStat).withColumn("mode", functions.lit("raw")) .unionByName(compareWithTest(filteredStat).withColumn("mode", functions.lit("filtered"))) 

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

हम जिस चार्ट में रुचि रखते हैं, उसके विनिर्देश इस तरह दिखेंगे:

 vegas.Vegas(width = 1024, height = 648) //   .withDataFrame(comparison.na.fill(0.0)) //           .encodeX("meanDrift", Quant, scale = Scale(domainValues = List(-1.0, 1.0), clamp = true)) //   -       .encodeY("train_fill", Quant) //       .encodeColor("outlieres", Quant, scale=Scale( rangeNominals=List("#00FF00", "#FF0000"), domainValues = List(0.0, 5), clamp = true)) //       .encodeSize("skewenes", Quant) //   -   (   ) .encodeShape("mode", Nom) .mark(vegas.Point) .show 

नीचे दिए गए चार्ट को इस तरह पढ़ना चाहिए:

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

तो, हम निम्नलिखित निष्कर्ष निकाल सकते हैं:

• कुछ संकेतों को एक उत्सर्जन फिल्टर की आवश्यकता होती है - हम 90 वें प्रतिशत के लिए अधिकतम मूल्यों को सीमित करेंगे।
• कुछ संकेत घातांक के करीब वितरण दर्शाते हैं - हम लघुगणक लेंगे।
• कुछ विशेषताएं परीक्षण में प्रस्तुत नहीं की जाती हैं - हम उन्हें प्रशिक्षण से बाहर कर देंगे।

सहसंबंध विश्लेषण

एक सामान्य विचार प्राप्त करने के बाद कि कैसे विशेषताओं को वितरित किया जाता है और वे प्रशिक्षण और परीक्षण सेट के बीच कैसे संबंधित हैं, आइए सहसंबंधों का विश्लेषण करने का प्रयास करें। ऐसा करने के लिए, पिछले अवलोकनों के आधार पर फीचर एक्सट्रैक्टर को कॉन्फ़िगर करें:

 //             val expressions = filteredTrain.schema.fieldNames //          .filterNot(x => x == "date" || x == "audit_experiment" || idsColumns(x) || x.contains("vd_")) .map(x => if(skewedFeautres(x)) { //      s"log($x) AS $x" } else { //     cappedFeatures.get(x).map(capping => s"IF($x < $capping, $x, $capping) AS $x").getOrElse(x) }) val rawFeaturesExtractor = new Pipeline().setStages(Array( new SQLTransformer().setStatement(s"SELECT ${expressions.mkString(", ")} FROM __THIS__"), new NullToDefaultReplacer(), new AutoAssembler().setOutputCol("features") )) //       val raw = rawFeaturesExtractor.fit(filteredTrain).transform( filteredTrain.where(isTestSimilar($"instanceId_userId"))) 

इस पाइपलाइन में नई मशीनरी में, SQLTransformer उपयोगिता ध्यान आकर्षित करती है, जो इनपुट तालिका के मनमाने ढंग से SQL परिवर्तनों की अनुमति देती है।

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

 val attributes = AttributeGroup.fromStructField(raw.schema("features")).attributes.get val originMap = filteredTrain .schema.filter(_.dataType == StringType) .flatMap(x => attributes.map(_.name.get).filter(_.startsWith(x.name + "_")).map(_ -> x.name)) .toMap //   ,          val isNonTrivialCorrelation = sqlContext.udf.register("isNonTrivialCorrelation", (x: String, y : String) => //    Scala-quiz   Option originMap.get(x).map(_ != originMap.getOrElse(y, "")).getOrElse(true)) 

वेक्टर कॉलम के साथ एक डेटासेट पर हाथ रखने से, स्पार्क का उपयोग करके क्रॉस-सहसंबंधों की गणना करना काफी सरल है, लेकिन परिणाम एक मैट्रिक्स है, जिसकी तैनाती के लिए आपको जोड़े के एक सेट में थोड़ा सा खेलना होगा:

 val pearsonCorrelation = //    Pearson  Spearman Correlation.corr(raw, "features", "pearson").rdd.flatMap( //           _.getAs[Matrix](0).rowIter.zipWithIndex.flatMap(x => { //   ,   (  , //  ) val name = attributes(x._2).name.get //    ,     x._1.toArray.zip(attributes).map(y => (name, y._2.name.get, y._1)) } //     DataFrame )).toDF("feature1", "feature2", "corr") .na.drop //   .where(isNonTrivialCorrelation($"feature1", $"feature2")) //    . pearsonCorrelation.coalesce(1).write.mode("overwrite") .parquet("sna2019/pearsonCorrelation") 

और, ज़ाहिर है, विज़ुअलाइज़ेशन: हमें फिर से गर्मी का नक्शा खींचने के लिए वेगास की मदद की ज़रूरत होगी:

 vegas.Vegas("Pearson correlation heatmap") .withDataFrame(pearsonCorrelation .withColumn("isPositive", $"corr" > 0) .withColumn("abs_corr", functions.abs($"corr")) .where("feature1 < feature2 AND abs_corr > 0.05") .orderBy("feature1", "feature2")) .encodeX("feature1", Nom) .encodeY("feature2", Nom) .encodeColor("abs_corr", Quant, scale=Scale(rangeNominals=List("#FFFFFF", "#FF0000"))) .encodeShape("isPositive", Nom) .mark(vegas.Point) .show 

परिणाम Zepl-e में देखने के लिए बेहतर है। एक सामान्य समझ के लिए:

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

 //   (GrpahX   ID) val featureIndexMap = spearmanCorrelation.select("feature1").distinct.rdd.map( _.getString(0)).collect.zipWithIndex.toMap val featureIndex = sqlContext.udf.register("featureIndex", (x: String) => featureIndexMap(x)) //    val vertices = sc.parallelize(featureIndexMap.map(x => x._2.toLong -> x._1).toSeq, 1) //    val edges = spearmanCorrelation.select(featureIndex($"feature1"), featureIndex($"feature2"), $"corr") //     .where("ABS(corr) > 0.7") .rdd.map(r => Edge(r.getInt(0), r.getInt(1), r.getDouble(2))) //       val components = Graph(vertices, edges).connectedComponents() val reversedMap = featureIndexMap.map(_.swap) //    ,    ,   //   val clusters = components .vertices.map(x => reversedMap(x._2.toInt) -> reversedMap(x._1.toInt)) .groupByKey().map(x => x._2.toSeq) .filter(_.size > 1) .sortBy(-_.size) .collect 

परिणाम तालिका के रूप में प्रस्तुत किया गया है:

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

मशीन सीखने

हम सबसे दिलचस्प बात करते हैं - मशीन लर्निंग। SparkML और PravdaML एक्सटेंशन का उपयोग करके सरलतम मॉडल (लॉजिस्टिक रिग्रेशन) का प्रशिक्षण देने की पाइपलाइन इस प्रकार है:

  new Pipeline().setStages(Array( new SQLTransformer().setStatement( """SELECT *, IF(array_contains(feedback, 'Liked'), 1.0, 0.0) AS label FROM __THIS__"""), new NullToDefaultReplacer(), new AutoAssembler() .setColumnsToExclude("date", "instanceId_userId", "instanceId_objectId", "feedback", "label") .setOutputCol("features"), Scaler.scale(Interceptor.intercept(UnwrappedStage.repartition( new LogisticRegressionLBFSG(), numPartitions = 127))) 

यहां हम न केवल कई परिचित तत्वों को देखते हैं, बल्कि कई नए भी हैं:

• LogisticRegressionLBFSG एक अनुमानक है जो लॉजिस्टिक प्रतिगमन के वितरित प्रशिक्षण के साथ है।
• वितरित एमएल-एल्गोरिदम से अधिकतम प्रदर्शन प्राप्त करने के लिए। विभाजन के दौरान डेटा को बेहतर तरीके से वितरित किया जाना चाहिए। UnwrappedStage.repartition उपयोगिता इसमें मदद करेगी, पाइप लाइन के लिए एक रिपार्टिशन ऑपरेशन को इस तरह से जोड़ना कि इसका उपयोग केवल प्रशिक्षण चरण में किया जाता है (आखिरकार, जब पूर्वानुमान का निर्माण होता है, तो यह अब आवश्यक नहीं है)।
• ताकि रैखिक मॉडल एक अच्छा परिणाम दे सके। डेटा को स्केल किया जाना चाहिए, जिसके लिए Scaler.scale उपयोगिता जिम्मेदार है। हालांकि, दो लगातार रैखिक परिवर्तनों (प्रतिगमन भार द्वारा स्केलिंग और गुणा) की उपस्थिति अनावश्यक खर्चों की ओर ले जाती है, और इन ऑपरेशनों को ध्वस्त करना वांछनीय है। PravdaML का उपयोग करते समय, आउटपुट एक परिवर्तन के साथ एक साफ मॉडल होगा :)।
• ठीक है, ज़ाहिर है, ऐसे मॉडलों के लिए आपको एक स्वतंत्र सदस्य की आवश्यकता होती है, जिसे हम इंटरसेप्टर.इन्टरसेप्ट ऑपरेशन का उपयोग करके जोड़ते हैं।

परिणामी पाइपलाइन, सभी डेटा पर लागू होती है, प्रति-उपयोगकर्ता AUC 0.6889 (सत्यापन कोड Zepl पर उपलब्ध है) देती है। अब यह हमारे सभी शोधों को लागू करने के लिए बना हुआ है: डेटा फ़िल्टर करें, सुविधाएँ बदलें, और खंड मॉडल। अंतिम पाइपलाइन इस तरह दिखाई देगी:

  new Pipeline().setStages(Array( new SQLTransformer().setStatement(s"SELECT instanceId_userId, instanceId_objectId, ${expressions.mkString(", ")} FROM __THIS__"), new SQLTransformer().setStatement("""SELECT *, IF(array_contains(feedback, 'Liked'), 1.0, 0.0) AS label, concat(IF(membership_status = 'A', 'OwnGroup_', 'NonUser_'), instanceId_objectType) AS type FROM __THIS__"""), new NullToDefaultReplacer(), new AutoAssembler() .setColumnsToExclude("date", "instanceId_userId", "instanceId_objectId", "feedback", "label", "type","instanceId_objectType") .setOutputCol("features"), CombinedModel.perType( Scaler.scale(Interceptor.intercept(UnwrappedStage.repartition( new LogisticRegressionLBFSG(), numPartitions = 127))), numThreads = 6) )) 

PravdaML — CombinedModel.perType. , numThreads = 6. .

, , per-user AUC 0.7004. ? , " " XGBoost :

 new Pipeline().setStages(Array( new SQLTransformer().setStatement("""SELECT *, IF(array_contains(feedback, 'Liked'), 1.0, 0.0) AS label FROM __THIS__"""), new NullToDefaultReplacer(), new AutoAssembler() .setColumnsToExclude("date", "instanceId_userId", "instanceId_objectId", "feedback", "label") .setOutputCol("features"), new XGBoostRegressor() .setNumRounds(100) .setMaxDepth(15) .setObjective("reg:logistic") .setNumWorkers(17) .setNthread(4) .setTrackerConf(600000L, "scala") )) 

, — XGBoost Spark ! DLMC , PravdaML , ( ). XGboost " " 10 per-user AUC 0.6981.

, , , . SparkML , . PravdaML : Parquet Spark:

 //     val perTypeWeights = sqlContext.read.parquet("sna2019/perType/stages/*/weights") //     20    ( //  ) val topFeatures = new TopKTransformer[Double]() .setGroupByColumns("type") .setColumnToOrderGroupsBy("abs_weight") .setTopK(20) .transform(perTypeWeights.withColumn("abs_weight", functions.abs($"unscaled_weight"))) .orderBy("type", "unscaled_weight") 

Parquet, PravdaML — TopKTransformer, .

Vegas ( Zepl ):

, - . XGBoost?

 val significance = sqlContext.read.parquet( "sna2019/xgBoost15_100_raw/stages/*/featuresSignificance" vegas.Vegas() .withDataFrame(significance.na.drop.orderBy($"significance".desc).limit(40)) .encodeX("name", Nom, sortField = Sort("significance", AggOps.Mean)) .encodeY("significance", Quant) .mark(vegas.Bar) .show 

, , XGBoost, , . . , XGBoost , , .

निष्कर्ष

, :). :

1. , Scala Spark , , , , .
2. Scala Spark Python: ETL ML, , , .
3. , , , (, ) , , .
4. , , . , , , -, .

, , , , -. , , " Scala " Newprolab.

, , — SNA Hackathon 2019 .