इस लेख को कैसे पढ़ें : मैं इस तथ्य के लिए माफी मांगता हूं कि पाठ इतना लंबा और अराजक निकला। आपके समय को बचाने के लिए, मैं प्रत्येक अध्याय की शुरुआत "व्हाट आई लर्न" के साथ करता हूं, जिसमें मैं एक या दो वाक्यों में अध्याय का सार समझाता हूं।

"बस समाधान दिखाओ!" यदि आप केवल यह देखना चाहते हैं कि मैं क्या कर रहा हूं, तो "और अधिक आविष्कारशील बनें" अध्याय पर जाएं, लेकिन मुझे लगता है कि विफलताओं के बारे में पढ़ना अधिक दिलचस्प और उपयोगी है।

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

सबसे पहले, कुछ परिचयात्मक स्पष्टीकरण।

डेटा

हमारे विश्वविद्यालय आनुवंशिक सूचना प्रसंस्करण केंद्र ने हमें टीएसवी डेटा के 25 टीबी प्रदान किए हैं। मैंने उन्हें Gzip द्वारा संपीड़ित 5 पैकेजों में तोड़ दिया, जिनमें से प्रत्येक में लगभग 240 चार-गीगाबाइट फाइलें थीं। प्रत्येक पंक्ति में एक व्यक्ति के एक एसएनपी के लिए डेटा होता था। कुल में, ~ 2.5 मिलियन एसएनपी और ~ 60 हजार लोगों पर डेटा प्रेषित किया गया था। एसएनपी जानकारी के अलावा, फाइलों में कई कॉलम थे, जिनमें विभिन्न विशेषताओं को दर्शाया गया था, जैसे कि रीडिंग इंटेंसिटी, विभिन्न एपल्स की आवृत्ति, आदि। अद्वितीय मानों के साथ लगभग 30 स्तंभ थे।

लक्ष्य

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

कैसे नहीं करना है

मैं एक उपयुक्त क्लिच का उद्धरण दूंगा:

मैं एक हज़ार बार असफल नहीं हुआ, मैंने सिर्फ एक हज़ार तरीके खोजे, जो प्रश्नों के लिए सुविधाजनक प्रारूप में डेटा का एक गुच्छा पार्स न करें।

पहला प्रयास

मैंने क्या सीखा : एक समय में 25 टीबी को पार्स करने का कोई सस्ता तरीका नहीं है।

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

एथेना को अपना डेटा और उसका प्रारूप दिखाने के बाद, मैंने कुछ प्रश्नों के साथ कुछ परीक्षण किए:

select * from intensityData limit 10;

और जल्दी से अच्छी तरह से संरचित परिणाम मिला। हो गया।

जब तक हमने काम में डेटा का उपयोग करने की कोशिश की ...

मुझे मॉडल पर परीक्षण करने के लिए सभी एसएनपी जानकारी खींचने के लिए कहा गया था। मैंने एक क्वेरी चलाई:

 select * from intensityData where snp = 'rs123456';

... और इंतजार किया। आठ मिनट और अनुरोधित डेटा के 4 से अधिक टीबी के बाद, मुझे परिणाम मिला। एथेना $ 5 प्रति टेराबाइट पर मिलने वाले डेटा की राशि के लिए शुल्क लेता है। इसलिए इस एकल अनुरोध में $ 20 और आठ मिनट प्रतीक्षा की लागत थी। सभी आंकड़ों के अनुसार मॉडल को चलाने के लिए, 38 साल तक इंतजार करना और $ 50 मिलियन का भुगतान करना आवश्यक था। जाहिर है, यह हमें सूट नहीं करता था।

Parquet का उपयोग करना आवश्यक था ...

मैंने क्या सीखा : अपनी Parquet फ़ाइलों और उनके संगठन के आकार से सावधान रहें।

सबसे पहले मैंने सभी TSV को Parquet फ़ाइलों में परिवर्तित करके स्थिति को ठीक करने का प्रयास किया। वे बड़े डेटा सेट के साथ काम करने के लिए सुविधाजनक हैं, क्योंकि उनमें जानकारी स्तंभ के रूप में संग्रहीत होती है: प्रत्येक स्तंभ अपनी स्वयं की मेमोरी / डिस्क सेगमेंट में निहित होता है, पाठ फ़ाइलों के विपरीत जिसमें पंक्तियों में प्रत्येक स्तंभ के तत्व होते हैं। और अगर आपको कुछ खोजने की आवश्यकता है, तो बस आवश्यक कॉलम पढ़ें। इसके अलावा, प्रत्येक फ़ाइल में मानों की एक श्रेणी को एक कॉलम में संग्रहीत किया जाता है, इसलिए यदि वांछित मान स्तंभ श्रेणी में नहीं है, तो स्पार्क समय बर्बाद नहीं करेगा पूरी फ़ाइल को स्कैन करना।

मैंने अपने TSVs को Parquet में बदलने के लिए एक साधारण AWS Glue कार्य चलाया और नई फ़ाइलों को एथेना में गिरा दिया। इसमें लगभग 5 घंटे लगे। लेकिन जब मैंने अनुरोध लॉन्च किया, तो इसे पूरा करने में लगभग एक ही समय और थोड़ा कम पैसा लगा। तथ्य यह है कि स्पार्क, कार्य को अनुकूलित करने की कोशिश कर रहा है, बस एक टीएसवी-चंक को अनपैक किया गया है और इसे अपने स्वयं के पर्किट-चंक में डाल दिया है। और चूंकि प्रत्येक चंक काफी बड़ा था और इसमें कई लोगों के पूर्ण रिकॉर्ड थे, इसलिए सभी एसएनपी प्रत्येक फ़ाइल में संग्रहीत किए गए थे, इसलिए स्पार्क को आवश्यक जानकारी निकालने के लिए सभी फाइलों को खोलना पड़ा।

उत्सुकतावश, Parquet में डिफ़ॉल्ट (और अनुशंसित) कंप्रेशन प्रकार - snappy - splitable नहीं है। इसलिए, प्रत्येक निष्पादक पूरे 3.5 जीबी डेटासेट को अनपैक करने और डाउनलोड करने के कार्य से जुड़ा हुआ है।

हम समस्या को समझते हैं

मैंने जो सीखा है : छंटाई मुश्किल है, खासकर अगर डेटा वितरित किया गया है।

ऐसा लग रहा था कि अब मुझे समस्या का सार समझ में आया। मुझे बस इतना करना था कि डेटा एसएनपी कॉलम द्वारा छाँटा जाए, लोगों द्वारा नहीं। फिर कई एसएनपी को एक अलग डेटा चंक में संग्रहित किया जाएगा, और फिर पैराकिट स्मार्ट फ़ंक्शन "केवल तभी खुलेगा जब मूल्य सीमा में होगा" अपने सभी महिमा में प्रकट होगा। दुर्भाग्य से, एक क्लस्टर में बिखरी हुई अरबों पंक्तियों को छांटना एक कठिन काम साबित हुआ है।

AWS निश्चित रूप से "मैं एक अनुपस्थित दिमाग वाला छात्र हूँ" के कारण पैसे वापस नहीं करना चाहता। जब मैंने अमेज़ॅन ग्लू पर छंटनी शुरू की, तो यह 2 दिनों के लिए काम किया और दुर्घटनाग्रस्त हो गया।

विभाजन के बारे में क्या?

मैंने क्या सीखा : स्पार्क में विभाजन संतुलित होना चाहिए।

फिर गुणसूत्रों पर डेटा को विभाजित करने के लिए मुझे विचार आया। उनमें से 23 हैं (और कुछ और, दिए गए माइटोकॉन्ड्रियल डीएनए और अनमैप्ड क्षेत्र)।
यह आपको डेटा को छोटे भागों में विभाजित करने की अनुमति देगा। यदि आप Glue स्क्रिप्ट में स्पार्क एक्सपोर्ट फ़ंक्शन में केवल एक partition_by = "chr" लाइन जोड़ते हैं, तो डेटा को बाल्टी में सॉर्ट किया जाना चाहिए।

जीनोम में कई टुकड़े होते हैं जिन्हें क्रोमोसोम कहा जाता है।

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

और अगर आप छोटे-छोटे विभाजन में भी विभाजित होते हैं?

मैंने क्या सीखा : कभी भी 2.5 मिलियन विभाजन करने की कोशिश न करें।

मैंने टहलने जाने का फैसला किया और हर एसएनपी का विभाजन किया। यह समान आकार के विभाजन की गारंटी देता है। बाड ए एन आईडिया था । मैंने ग्लू का फायदा उठाया और निर्दोष partition_by = 'snp' जोड़ partition_by = 'snp' । टास्क शुरू हुआ और चलने लगा। एक दिन बाद, मैंने जाँच की और देखा कि S3 में अब तक कुछ भी नहीं लिखा गया था, इसलिए मैंने इस कार्य को मार दिया। ऐसा लग रहा है कि ग्लू S3 में एक छिपी जगह पर इंटरमीडिएट फाइलें लिख रहा था, और बहुत सारी फाइलें, शायद कुछ लाखों। नतीजतन, मेरी गलती की कीमत एक हजार डॉलर से अधिक थी और मेरे गुरु को खुश नहीं किया।

विभाजन + छँटाई

मैंने क्या सीखा : स्पार्क को स्थापित करना अभी भी मुश्किल है।

विभाजन का अंतिम प्रयास यह था कि मैंने गुणसूत्रों का विभाजन किया और फिर प्रत्येक विभाजन को क्रमबद्ध किया। सिद्धांत रूप में, यह प्रत्येक अनुरोध को गति देगा, क्योंकि वांछित एसएनपी डेटा किसी दिए गए सीमा के भीतर कई Parquet विखंडू के भीतर होना चाहिए। काश, विभाजन किए गए डेटा को छांटना भी एक मुश्किल काम साबित होता। नतीजतन, मैंने एक कस्टम क्लस्टर के लिए EMR पर स्विच किया और एक अधिक लचीली वर्कफ़्लो बनाने के लिए आठ शक्तिशाली इंस्टेंस (C5.4xl) और स्पार्कलियर का उपयोग किया ...

 # Sparklyr snippet to partition by chr and sort w/in partition # Join the raw data with the snp bins raw_data group_by(chr) %>% arrange(Position) %>% Spark_write_Parquet( path = DUMP_LOC, mode = 'overwrite', partition_by = c('chr') )

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

मैं और अधिक आविष्कारशील हो रहा हूँ

मैंने क्या सीखा : कभी-कभी विशेष डेटा के लिए विशेष समाधान की आवश्यकता होती है।

प्रत्येक एसएनपी का एक स्थिति मूल्य होता है। यह गुणसूत्र के साथ स्थित आधारों की संख्या के अनुरूप संख्या है। यह हमारे डेटा को व्यवस्थित करने का एक अच्छा और प्राकृतिक तरीका है। सबसे पहले मैं प्रत्येक गुणसूत्र के क्षेत्र द्वारा विभाजन करना चाहता था। उदाहरण के लिए, पदों 1 - 2000, 2001 - 4000, आदि। लेकिन समस्या यह है कि एसएनपी समान रूप से गुणसूत्रों में वितरित नहीं किए जाते हैं, यही वजह है कि समूहों का आकार बहुत भिन्न होगा।

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

अंत में, मैंने 75 एसएनपी पर समूह (बिन) बनाए, मैं नीचे कारण बताऊंगा।

 snp_to_bin <- unique_snps %>% group_by(chr) %>% arrange(position) %>% mutate( rank = 1:n() bin = floor(rank/snps_per_bin) ) %>% ungroup()

पहले स्पार्क के साथ प्रयास करें

मैंने क्या सीखा : स्पार्क एकीकरण तेज है, लेकिन विभाजन अभी भी महंगा है।

मैं स्पार्क में इस छोटे (2.5 मिलियन लाइनों) डेटा फ्रेम को पढ़ना चाहता था, इसे कच्चे डेटा के साथ जोड़ रहा था, और फिर नए जोड़े गए bin कॉलम द्वारा विभाजन।

 # Join the raw data with the snp bins data_w_bin <- raw_data %>% left_join(sdf_broadcast(snp_to_bin), by ='snp_name') %>% group_by(chr_bin) %>% arrange(Position) %>% Spark_write_Parquet( path = DUMP_LOC, mode = 'overwrite', partition_by = c('chr_bin') )

मैंने sdf_broadcast() उपयोग किया, इसलिए स्पार्क को पता चला कि उसे सभी नोड्स को एक डेटा फ्रेम भेजना चाहिए। यह उपयोगी है यदि डेटा छोटा है और सभी कार्यों के लिए आवश्यक है। अन्यथा, स्पार्क स्मार्ट होने की कोशिश करता है और आवश्यकतानुसार डेटा वितरित करता है, जो ब्रेक का कारण बन सकता है।

और फिर से, मेरे विचार से काम नहीं हुआ: कार्यों ने थोड़ी देर के लिए काम किया, विलय को पूरा किया, और फिर, विभाजन द्वारा निष्पादित निष्पादकों की तरह, वे असफल होने लगे।

AWK जोड़ें

मैंने क्या सीखा : जब मूल बातें आपको सिखाती हैं तो नींद नहीं आती। 1980 के दशक में निश्चित रूप से किसी ने आपकी समस्या को पहले ही हल कर दिया था।

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

मैंने स्तंभ मानों को कैसे तोड़ा जाए, इसके लिए स्टैकऑवरफ़्लो पर खोज की और ऐसा शानदार उत्तर मिला । AWK का उपयोग करके, आप परिणामों को भेजने के बजाय स्क्रिप्ट पर लिखकर कॉलम मानों में एक टेक्स्ट फ़ाइल को विभाजित कर सकते हैं।

परीक्षण के लिए, मैंने एक बैश स्क्रिप्ट लिखी। मैंने पैक किए गए TSV में से एक को डाउनलोड किया, फिर इसे गज़िप से पैक किया और इसे awk लिए भेज दिया।

 gzip -dc path/to/chunk/file.gz | awk -F '\t' \ '{print $1",..."$30">"chunked/"$chr"_chr"$15".csv"}'

यह काम किया!

कोर फिलिंग

मैंने क्या सीखा : gnu parallel एक जादुई चीज है, हर किसी को इसका उपयोग करना चाहिए।

जुदाई धीमी थी, और जब मैंने एक शक्तिशाली (और महंगी) EC2 उदाहरण के उपयोग का परीक्षण करने के लिए htop शुरू किया, तो यह पता चला कि मैं केवल एक कोर और लगभग 200 एमबी मेमोरी का उपयोग कर रहा था। समस्या को हल करने और बहुत सारे पैसे न खोने के लिए, यह पता लगाना आवश्यक था कि काम को कैसे समानांतर किया जाए। सौभाग्य से, कमांड लाइन पुस्तक में जेरोन जेंसेंस के आश्चर्यजनक डेटा विज्ञान में , मुझे समानांतरकरण पर एक अध्याय मिला। इससे मैंने gnu parallel बारे में सीखा, यूनिक्स पर मल्टीथ्रेडिंग को लागू करने के लिए एक बहुत ही लचीला तरीका।

जब मैंने एक नई प्रक्रिया का उपयोग करके विभाजन शुरू किया, तो सब कुछ ठीक था, लेकिन एक अड़चन थी - डिस्क में S3 ऑब्जेक्ट डाउनलोड करना बहुत तेज़ नहीं था और पूरी तरह से समानांतर नहीं था। इसे ठीक करने के लिए, मैंने यह किया:

मुझे पता चला कि पाइप लाइन में सीधे एस 3-डाउनलोड कदम को लागू करना संभव है, डिस्क पर मध्यवर्ती भंडारण को पूरी तरह से समाप्त कर देता है। इसका मतलब यह है कि मैं कच्चे डेटा को डिस्क पर लिखने और उससे भी छोटे उपयोग करने से बच सकता हूं, और इसलिए एडब्ल्यूएस पर सस्ता भंडारण।
aws configure set default.s3.max_concurrent_requests 50 कमांड कमांड aws configure set default.s3.max_concurrent_requests 50 ने थ्रेड्स की संख्या में वृद्धि की है जो AWS CLI का उपयोग करता है (डिफ़ॉल्ट रूप से 10 हैं)।
मैंने नाम के अक्षर n के साथ नेटवर्क गति के लिए अनुकूलित EC2 उदाहरण पर स्विच किया। मैंने पाया कि एन-इंस्टेंस का उपयोग करते समय कंप्यूटिंग शक्ति का नुकसान डाउनलोड गति में वृद्धि से ऑफसेट से अधिक है। अधिकांश कार्यों के लिए, मैंने c5n.4xl का उपयोग किया।
मैंने gzip को pigz बदल दिया, यह एक gzip उपकरण है जो फ़ाइलों को अनपैक करने के pigz को pigz करने के लिए शांत चीजें कर सकता है (इससे सबसे कम मदद मिली)।

 # Let S3 use as many threads as it wants aws configure set default.s3.max_concurrent_requests 50 for chunk_file in $(aws s3 ls $DATA_LOC | awk '{print $4}' | grep 'chr'$DESIRED_CHR'.csv') ; do aws s3 cp s3://$batch_loc$chunk_file - | pigz -dc | parallel --block 100M --pipe \ "awk -F '\t' '{print \$1\",...\"$30\">\"chunked/{#}_chr\"\$15\".csv\"}'" # Combine all the parallel process chunks to single files ls chunked/ | cut -d '_' -f 2 | sort -u | parallel 'cat chunked/*_{} | sort -k5 -n -S 80% -t, | aws s3 cp - '$s3_dest'/batch_'$batch_num'_{}' # Clean up intermediate data rm chunked/* done

इन चरणों को एक दूसरे के साथ जोड़ा जाता है ताकि सब कुछ बहुत तेज़ी से काम करे। बढ़ी हुई डाउनलोड गति और डिस्क पर लिखने की अस्वीकृति के लिए धन्यवाद, मैं अब कुछ ही घंटों में 5-टेराबाइट पैकेज को संसाधित कर सकता हूं।

इस ट्वीट में 'टीएसवी' का उल्लेख होना था। अफसोस।

पुन: पार्स डेटा का उपयोग करना

मैंने क्या सीखा : स्पार्क असम्पीडित डेटा को पसंद करता है और विभाजन को जोड़ना पसंद नहीं करता है।

अब डेटा S3 में एक अनपैक्ड (पढ़ें, साझा किए गए) और अर्ध-ऑर्डर किए गए प्रारूप में था, और मैं फिर से स्पार्क पर लौट सकता था। एक आश्चर्य मुझे इंतजार कर रहा था: मैं फिर से वांछित हासिल करने में विफल रहा! स्पार्क को यह बताना बहुत मुश्किल था कि डेटा का विभाजन कैसे हुआ। और जब मैंने ऐसा किया, तब भी यह पता चला कि बहुत अधिक विभाजन (95 हजार) थे, और जब मैंने उनकी संख्या को कम करने के लिए सुसंगत सीमा के साथ कम coalesce , तो इसने मेरे विभाजन को बर्बाद कर दिया। मुझे यकीन है कि इसे ठीक किया जा सकता है, लेकिन खोज के कुछ दिनों में, मुझे कोई समाधान नहीं मिला। अंत में, मैंने स्पार्क में सभी कार्यों को पूरा किया, हालांकि इसमें कुछ समय लगा, और मेरी विभाजन Parquet फाइलें बहुत छोटी नहीं थीं (~ 200 Kb)। हालाँकि, डेटा वह था जहाँ इसकी आवश्यकता थी।

बहुत छोटा और अलग, अद्भुत!

स्थानीय स्पार्क अनुरोधों का परीक्षण

मैंने क्या सीखा : सरल समस्याओं को हल करने में स्पार्क बहुत अधिक उपरि है।

स्मार्ट प्रारूप में डेटा डाउनलोड करके, मैं गति का परीक्षण करने में सक्षम था। मैंने स्थानीय स्पार्क सर्वर को शुरू करने के लिए आर पर एक स्क्रिप्ट स्थापित की, और फिर मैंने स्पार्क डेटा फ़्रेम को Parquet समूहों (बिन) के निर्दिष्ट भंडार से लोड किया। मैंने सभी डेटा लोड करने की कोशिश की, लेकिन विभाजन को पहचानने के लिए स्पार्कलियर नहीं मिला।

 sc <- Spark_connect(master = "local") desired_snp <- 'rs34771739' # Start a timer start_time <- Sys.time() # Load the desired bin into Spark intensity_data <- sc %>% Spark_read_Parquet( name = 'intensity_data', path = get_snp_location(desired_snp), memory = FALSE ) # Subset bin to snp and then collect to local test_subset <- intensity_data %>% filter(SNP_Name == desired_snp) %>% collect() print(Sys.time() - start_time)

निष्पादन में 29.415 सेकंड लगे। बहुत बेहतर है, लेकिन कुछ भी बड़े पैमाने पर परीक्षण के लिए अच्छा नहीं है। इसके अलावा, मैं कैशिंग के साथ काम को गति नहीं दे सकता था, क्योंकि जब मैंने मेमोरी में डेटा फ़्रेम को कैश करने की कोशिश की, स्पार्क हमेशा दुर्घटनाग्रस्त हो गया, तब भी जब मैंने 15 से कम वजन वाले डेटासेट के लिए 50 जीबी से अधिक मेमोरी आवंटित की।

AWK पर लौटें

मैंने क्या सीखा : AWK साहचर्य सरणियाँ बहुत कुशल हैं।

मैं समझ गया कि मैं एक उच्च गति प्राप्त कर सकता हूं। मुझे याद आया कि ब्रूस बार्नेट के उत्कृष्ट AWK गाइड में, मैंने " साहचर्य सरणियों " नामक एक अच्छी सुविधा के बारे में पढ़ा। वास्तव में, ये कुंजी-मूल्य जोड़े हैं, जिन्हें किसी कारण से AWK में अलग-अलग रूप से कहा जाता था, और इसलिए मैंने किसी भी तरह विशेष रूप से उनका उल्लेख नहीं किया। रोमन चेलिपाका ने याद किया कि "साहचर्य सरणियों" शब्द "की-वैल्यू पेयर" शब्द की तुलना में बहुत पुराना है। यदि आप Google Ngram में की-वैल्यू खोजते हैं , तो भी आपको यह शब्द दिखाई नहीं देगा, लेकिन आपको साहचर्य सरणियाँ मिलेंगी ! इसके अलावा, कुंजी-मूल्य जोड़ी अक्सर डेटाबेस से जुड़ी होती है, इसलिए हैशमैप के साथ तुलना करना बहुत अधिक तर्कसंगत है। मुझे एहसास हुआ कि मैं अपने एसएनपी को बिन टेबल और कच्चे डेटा को स्पार्क का उपयोग किए बिना लिंक करने के लिए इन साहचर्य सरणियों का उपयोग कर सकता हूं।

इसके लिए, AWK स्क्रिप्ट में, मैंने BEGIN ब्लॉक का उपयोग किया। यह कोड का एक टुकड़ा है जो डेटा की पहली पंक्ति को स्क्रिप्ट के मुख्य निकाय में स्थानांतरित करने से पहले निष्पादित किया जाता है।

 join_data.awk BEGIN { FS=","; batch_num=substr(chunk,7,1); chunk_id=substr(chunk,15,2); while(getline < "snp_to_bin.csv") {bin[$1] = $2} } { print $0 > "chunked/chr_"chr"_bin_"bin[$1]"_"batch_num"_"chunk_id".csv" }

while(getline...) कमांड ने CSV समूह (बिन) से सभी पंक्तियों को लोड किया, पहले स्तंभ (SNP नाम) को bin सहयोगी सरणी के लिए कुंजी के रूप में और दूसरे मान (समूह) को मान के रूप में सेट किया। फिर, ब्लॉक { } , जिसे मुख्य फ़ाइल की सभी लाइनों पर लागू किया जाता है, प्रत्येक लाइन आउटपुट फ़ाइल को भेजी जाती है, जिसे इसके समूह (बिन) के आधार पर एक विशिष्ट नाम मिलता है: ..._bin_"bin[$1]"_...

batch_num और chunk_id पाइप लाइन द्वारा उपलब्ध कराए गए डेटा के अनुरूप थे, जो कि रेस स्टेट से बचा था, और parallel द्वारा लॉन्च किए गए प्रत्येक निष्पादन थ्रेड को अपनी अनूठी फ़ाइल में लिखा गया था।

चूंकि मैंने AWK के साथ अपने पिछले प्रयोग के बाद छोड़े गए गुणसूत्रों पर सभी कच्चे डेटा को फ़ोल्डरों में बिखेर दिया था, अब मैं एक बार में गुणसूत्र पर प्रक्रिया करने के लिए एक और बैश स्क्रिप्ट लिख सकता था और S3 को गहरा विभाजन डेटा दे सकता था।

 DESIRED_CHR='13' # Download chromosome data from s3 and split into bins aws s3 ls $DATA_LOC | awk '{print $4}' | grep 'chr'$DESIRED_CHR'.csv' | parallel "echo 'reading {}'; aws s3 cp "$DATA_LOC"{} - | awk -v chr=\""$DESIRED_CHR"\" -v chunk=\"{}\" -f split_on_chr_bin.awk" # Combine all the parallel process chunks to single files and upload to rds using R ls chunked/ | cut -d '_' -f 4 | sort -u | parallel "echo 'zipping bin {}'; cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R '$S3_DEST'/chr_'$DESIRED_CHR'_bin_{}.rds" rm chunked/*

स्क्रिप्ट के दो खंड parallel ।

पहला खंड वांछित गुणसूत्र की जानकारी वाली सभी फाइलों के डेटा को पढ़ता है, फिर इस डेटा को स्ट्रीम में वितरित किया जाता है जो फाइलों को संबंधित समूहों (बिन) में बिखेरता है। दौड़ की स्थितियों को होने से रोकने के लिए जब एक ही फ़ाइल में कई धाराएँ लिखी जाती हैं, तो AWK फ़ाइल नामों को विभिन्न स्थानों पर डेटा लिखने के लिए स्थानांतरित करता है, उदाहरण के लिए, chr_10_bin_52_batch_2_aa.csv । नतीजतन, डिस्क पर कई छोटी फाइलें बनाई जाती हैं (इसके लिए मैंने टेराबाइट ईबीएस वॉल्यूम का उपयोग किया है)।

दूसरे parallel खंड से पाइप लाइन समूहों (बिन) के माध्यम से जाती है और cat साथ आम सीएसवी में उनकी व्यक्तिगत फ़ाइलों को जोड़ती है, और फिर उन्हें निर्यात के लिए भेजती है।

आर के लिए प्रसारण?

मैंने क्या सीखा : आप एक आर स्क्रिप्ट से stdin और stdout उपयोग कर सकते हैं, और इसलिए इसे पाइपलाइन में उपयोग करें।

बैश स्क्रिप्ट में, आप इस लाइन को नोटिस कर सकते हैं: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R... ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R... यह सभी संक्षिप्त समूह फ़ाइलों (बिन) का अनुवाद नीचे की आर स्क्रिप्ट में करता है। {} एक विशेष parallel तकनीक है जो इसके द्वारा भेजे गए किसी भी डेटा को सीधे निर्दिष्ट कमांड में ही डालती है। {#} विकल्प एक अद्वितीय थ्रेड आईडी प्रदान करता है, और {%} नौकरी स्लॉट संख्या (बार-बार, लेकिन कभी भी एक साथ नहीं {%} प्रतिनिधित्व करता है। प्रलेखन में सभी विकल्पों की एक सूची मिल सकती है ।

 #!/usr/bin/env Rscript library(readr) library(aws.s3) # Read first command line argument data_destination <- commandArgs(trailingOnly = TRUE)[1] data_cols <- list(SNP_Name = 'c', ...) s3saveRDS( read_csv( file("stdin"), col_names = names(data_cols), col_types = data_cols ), object = data_destination )

जब file("stdin") readr::read_csv file("stdin") चर को readr::read_csv , तो R-स्क्रिप्ट में अनुवादित डेटा को फ्रेम में लोड किया जाता है, जिसे बाद में सीधे .rds फ़ाइल में .rds रूप में S3 में लिखा जाता है।

RDS स्तंभ भंडारण के तामझाम के बिना, लकड़ी की छत के एक छोटे संस्करण की तरह है।

बैश स्क्रिप्ट को पूरा करने के बाद, मुझे S3 में पड़ी हुई .rds फ़ाइलों .rds एक .rds मिला, जिसने मुझे कुशल संपीड़न और अंतर्निहित प्रकारों का उपयोग करने की अनुमति दी।

ब्रेक आर का उपयोग करने के बावजूद, सब कुछ बहुत तेजी से काम किया। यह आश्चर्य की बात नहीं है कि डेटा पर पढ़ने और लिखने के लिए जिम्मेदार आर पर टुकड़े अच्छी तरह से अनुकूलित हैं। एक मध्यम आकार के गुणसूत्र पर परीक्षण के बाद, कार्य को C5n.4xl उदाहरण पर लगभग दो घंटे में पूरा किया गया।

S3 की सीमाएँ

मैंने क्या सीखा : पथ के स्मार्ट कार्यान्वयन के लिए धन्यवाद, S3 कई फ़ाइलों को संसाधित कर सकता है।

मुझे चिंता थी कि अगर S3 इसमें स्थानांतरित की गई बहुत सारी फ़ाइलों को संभाल सकता है। मैं फ़ाइल नामों को सार्थक बना सकता था, लेकिन S3 उनके लिए कैसा दिखेगा?

S3 में फ़ोल्डर सिर्फ सुंदरता के लिए हैं, वास्तव में, सिस्टम को प्रतीक में दिलचस्पी नहीं है /। S3 FAQ पृष्ठ से।

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

Amazon, , «-----» . : get-, . , 20 . bin-. , , (, , ). .

?

: — .

: « ?» ( gzip CSV- 7 ) . , R Parquet ( Arrow) Spark. R, , , .

: , .

, .
EC2 , ( , Spark ). , , AWS- 10 .

R .

S3 , .

 library(aws.s3) library(tidyverse) chr_sizes <- get_bucket_df( bucket = '...', prefix = '...', max = Inf ) %>% mutate(Size = as.numeric(Size)) %>% filter(Size != 0) %>% mutate( # Extract chromosome from the file name chr = str_extract(Key, 'chr.{1,4}\\.csv') %>% str_remove_all('chr|\\.csv') ) %>% group_by(chr) %>% summarise(total_size = sum(Size)/1e+9) # Divide to get value in GB # A tibble: 27 x 2 chr total_size <chr> <dbl> 1 0 163. 2 1 967. 3 10 541. 4 11 611. 5 12 542. 6 13 364. 7 14 375. 8 15 372. 9 16 434. 10 17 443. # … with 17 more rows

, , , num_jobs , .

 num_jobs <- 7 # How big would each job be if perfectly split? job_size <- sum(chr_sizes$total_size)/7 shuffle_job <- function(i){ chr_sizes %>% sample_frac() %>% mutate( cum_size = cumsum(total_size), job_num = ceiling(cum_size/job_size) ) %>% group_by(job_num) %>% summarise( job_chrs = paste(chr, collapse = ','), total_job_size = sum(total_size) ) %>% mutate(sd = sd(total_job_size)) %>% nest(-sd) } shuffle_job(1) # A tibble: 1 x 2 sd data <dbl> <list> 1 153. <tibble [7 × 3]>

purrr .

 1:1000 %>% map_df(shuffle_job) %>% filter(sd == min(sd)) %>% pull(data) %>% pluck(1)

, . Bash- for . 10 . , . , .

 for DESIRED_CHR in "16" "9" "7" "21" "MT" do # Code for processing a single chromosome fi

 sudo shutdown -h now

… ! AWS CLI user_data Bash- . , .

 aws ec2 run-instances ...\ --tag-specifications "ResourceType=instance,Tags=[{Key=Name,Value=<<job_name>>}]" \ --user-data file://<<job_script_loc>>

!

: API .

- . , . API . .rds Parquet-, , . R-.

, , get_snp . pkgdown , .

: , !

SNP , (binning) . SNP, (bin). ( ) .

 # Part of get_snp() ... # Test if our current snp data has the desired snp. already_have_snp <- desired_snp %in% prev_snp_results$snps_in_bin if(!already_have_snp){ # Grab info on the bin of the desired snp snp_results <- get_snp_bin(desired_snp) # Download the snp's bin data snp_results$bin_data <- aws.s3::s3readRDS(object = snp_results$data_loc) } else { # The previous snp data contained the right bin so just use it snp_results <- prev_snp_results } ...

, . , . , dplyr::filter , , .

, prev_snp_results snps_in_bin . SNP (bin), , . SNP (bin) :

 # Get bin-mates snps_in_bin <- my_snp_results$snps_in_bin for(current_snp in snps_in_bin){ my_snp_results <- get_snp(current_snp, my_snp_results) # Do something with results }

( ) , . , . .

, , , …

. . ( ), , (bin) , SNP 0,1 , , S3 .

निष्कर्ष

— . , . , . , , , . , , , , . , , , , - .

. , , «» , . .

:

25 ;
Parquet- ;
Spark ;
2,5 ;
, Spark;
;
Spark , ;
, , - 1980-;
gnu parallel — , ;
Spark ;
Spark ;
AWK ;
stdin stdout R-, ;
S3 ;
— ;
, ;
API ;
, !

AWK और R के साथ परसिम 25TB