Kuidas seda artiklit lugeda: vabandan, et tekst osutus nii pikaks ja segaseks. Aja säästmiseks alustan igat peatükki sissejuhatusega 'Mida ma õppisin', milles kirjeldan peatüki sisu ühes-kahes lauses.

„Lihtsalt näita lahendust!“ Kui soovite lihtsalt näha, millega ma jõudsin, minge peatükki 'Muudan ennast leidlikumaks', kuid arvan, et huvitavam ja kasulikum on lugeda ebaõnnestumistest.

Hiljuti paluti mul seadistada protsess suure hulga DNA algsete järjestuste töötlemiseks (tehniliselt on see SNP-kiip). Oluline oli kiiresti saada andmeid konkreetse geneetilise asukoha (mida nimetatakse SNP-ks) kohta edasise modelleerimise ja muude ülesannete jaoks. Kasutades R-i ja AWK-d, suutsin andmed looduslikul viisil puhastada ja korraldada, suurendades märgatavalt päringute töötlemise kiirus. See polnud lihtne ja nõudis mitmeid iteratsioone. See artikkel aitab teil vältida mõningaid minu vigu ja demonstreerib, millega ma lõpuks välja jõudsin.

Alustuseks mõned sissejuhatused.

Andmed

Meie ülikooli geneetilise informatsiooni töötlemise keskus andis meile andmed TSV formaadis, mille maht oli 25 TB. Need andmed tulid minuni jagatuna 5 paketti, Gzip-iga kompresseeritult, millest igaühes oli ligikaudu 240 nelja gigabaiti suurt faili. Iga rida sisaldas andmeid ühe inimese üksiku SNP kohta. Kokku edastati andmed ~2,5 miljoni SNP ja ~60 tuhande inimese kohta. Peale SNP-informatsiooni sisaldasid failid mitmeid veerge numbritega, mis kajastasid erinevaid omadusi, nagu lugemise intensiivsus, erinevate alleelide sagedus jne. Kokku oli umbes 30 veergu unikaalsete väärtustega.

Eesmärk

Nagu igasugustes andmehaldusprojektides, oli kõige olulisem määrata, kuidas andmeid kasutatakse. Antud juhul me peamiselt kohandame mudeleid ja töökäike SNP-de jaoks SNP-de põhjal. See tähendab, et me vajame andmeid samaaegselt ainult ühe SNP kohta. Ma pidin õppima, kuidas võimalikult lihtsalt, kiiresti ja odavalt kõik salvestused ühe 2,5 miljoni SNP kohta välja tuua.

Kuidas seda mitte teha

Tsiteerin sobivat klišeed:

Ma ei ole tuhat korda ebaõnnestunud, olen lihtsalt avastanud tuhat viisi, kuidas andmeid ei saa hõlpsalt analüüsi jaoks korraldada.

Esimene katse

Mida ma õppisin: ei ole odavat viisi 25 TB andmete korraga analüüsimiseks.

Kuulates Vanderbilti ülikoolis loengut „Suure andmekoguse töötluse edasised meetodid“, olin kindel, et asi on selge. Tõenäoliselt kulub tund või kaks Hive-serveri seadistamiseks, et kõiki andmeid läbi käia ja tulemusi esitada. Kuna meie andmed on salvestatud AWS S3-sse, kasutasin teenust Athena, mis võimaldab rakendada Hive SQL-päringuid S3 andmete jaoks. Hive-klastri seadistama ei pea ja maksad ainult nende andmete eest, mida otsid.

Pärast seda, kui näitasin Athenale oma andmeid ja nende vormingut, käivitasin mitu testi sarnaste päringutega:

select * from intensityData limit 10;

Ja sain kiiresti hästi struktureeritud tulemused. Valmis.

Kuni me ei olnud proovinud andmeid töös…

Küsimisel paluti mul välja tuua kogu teave SNP-de kohta, et testida mudelit selle peal. Käivitasin päringu:

select * from intensityData 
where snp = 'rs123456';

…ja hakkasin ootama. Pärast kaheksa minuti ja rohkem kui 4 TB küsitud andmete ootamist, sain tulemuse. Athena maksab leitud andmete mahu eest, $5 TB kohta. Nii et see ainus päring maksis $20 ja ooteaeg oli kaheksa minutit. Et mudelit kõigi andmete peal rakendada, pidi ootama 38 aastat ja maksma $50 miljonit. Ilmselgelt see ei sobinud meile.

Tuli kasutada Parquet…

Mida ma õppisin: olge ettevaatlik oma Parquet-failide suuruse ja nende korralduse osas.

Alguses proovisin olukorda parandada, muutes kõik TSV-d Parquet-failideks. Need on suured andmekogumitega töötamiseks mugavad, kuna teave salvestatakse neis veergude kaupa: iga veerg asub oma mälu/diskisektorites, erinevalt tekstifailidest, kus read sisaldavad iga veeru elemente. Ja kui midagi tuleb leida, piisab vajalikku veergu lugeda. Lisaks hoitakse igas failis veerus väärtuste vahemikku, seega kui otsitav väärtus ei asu veeru vahemikus, ei raiska Spark aega kogu faili skannimisele.

Käivitasin lihtsa ülesande AWS Glue et meie TSV-d Parquet'iks konverteerida ja laadisin uusi faile Athene. See võttis umbes 5 tundi. Kuid kui ma jooksutasin päringu, siis selle täitmiseks kulus umbes sama palju aega ja veidi vähem raha. Asi on selles, et Spark, püüdes ülesannet optimeerida, lihtsalt lahti pakkis ühe TSV-chunki ja pani selle oma Parquet-chunki. Ja kuna iga chunk oli piisavalt suur ning sisaldas palju inimesi, oli igas failis kõik SNP-d, mistõttu pidi Spark avama kõik failid, et vajalikke andmeid kätte saada.

Küll on huvitav, et Parquet'i vaike- (ja soovitatud) kompressioonitüüp — snappy — ei ole jaotatav (splitable). Seetõttu kinni kõik täitjad (executor) ülesande lahti pakkimisel ja 3,5 GB täisdataseti üleslaadimisel.

Selgitame probleemi

Mida ma õppisin: sorteerimine on keeruline, eriti kui andmed on jaotatud.

Ma arvasin, et nüüd saan probleemi olemusest aru. Ma pidin lihtsalt sorteerima andmed SNP veeru järgi, mitte inimeste järgi. Siis oleks ühes andmechunkis mitu SNP-d ja siis läheks Parquet'i „nutikas” funktsioon „ava ainult juhul, kui väärtus on vahemikus” tõeliselt töösse. Kahjuks osutus miljardi rea sorteerimine, mis oli hajutatud klastrisse, keeruliseks ülesandeks.

Mina, kui võtan ülikoolis algoritmide tundi: «Uhh, kedagi ei huvita nende sorteerimisalgoritmide arvutuslik keerukus»

Mina, kui proovisin sorteerida 20TB #spark tabelis: «Miks see nii kaua aega võtab?» #DataScience raske aeg.

— Nick Strayer (@NicholasStrayer) 11. märts 2019

AWS ei taha kindlasti raha tagasi anda põhjusega «Ma olen hajameelne üliõpilane». Pärast seda, kui ma käivitasin sorteerimise Amazon Glue'is, töötas see 2 päeva ja kukkus kokku.

Aga mis on partitsioneerimisega?

Mida ma õppisin: Sparkis peavad partitsioonid olema tasakaalus.

Siis tuli mulle idee partitsioneerida andmed kromosoomide kaupa. Neid on 23 (ja veel mõned, kui arvestada mitoondria DNA-d ja lahtiseks jäävad (unmapped) alad).
See võimaldaks jagada andmed väiksemateks osadeks. Kui lisada Spark-funktsiooni eksportimise skriptis Glue vaid üks rida partition_by = "chr", siis oleks andmed jagatud baketideks (buckets).

Genoom koosneb paljusid fragmente, mida nimetatakse kromosoomideks.

Kahjuks see ei töötanud. Kromosoomidel on erinevad suurused, seega ka erinev teabe hulk. See tähendas, et ülesanded, mille Spark saatis töötajatele, ei olnud tasakaalus ja kestsid aeglaselt, kuna mõned sõlmed lõpetasid varakult ja seisisid ootel. Kuid ülesanded olid täidetud. Siiski, kui pärisin ühe SNP, tõi tasakaalustamatus jälle probleeme. SNP töötlemise kulu suuremates kromosoomides (st sealt, kust soovime andmeid saada) vähenes vaid umbes 10 korda. Palju, aga ikka ebapiisavalt.

Aga kui jagada veel väiksemateks partitsioonideks?

Mida ma õppisin: ärge kunagi proovige teha 2,5 miljonit partitsiooni.

Ma otsustasin minna edasi ja partitsioneerida iga SNP. See tagas partitsioonide sama suuruse. HALB IDEE. Kasutasin Glue'i ja lisasin süütu rida partition_by = 'snp'. Ülesanne käivitus ja hakkas töötama. Päeva pärast kontrollisin ja nägin, et S3-s pole ikka veel midagi kirjutatud, seega lõpetasin ülesande. Tundub, et Glue kirjutas vahefaile peidetud kohta S3-s, ja palju faile, võib-olla paar miljonit. Tulemuseks oli minu viga ulatuses üle tuhande dollari ja ei rõõmustanud mu mentorit.

Partitsioneerimine + sorteerimine

Mida ma õppisin: sortimine on endiselt keeruline, nagu ka Spark'i konfigureerimine.

Viimane partitsioneerimise katse seisnes selles, et ma partitsioneerin kromosoome ja seejärel sorteerin iga partitsiooni. Teoorias peaks see kiirendama iga päringut, kuna soovitud SNP andmed peaksid olema mõnes Parquet-chunkis antud vahemikus. Kahjuks osutus isegi partitsioneeritud andmete sorteerimine keeruliseks ülesandeks. Lõpuks läksin EMR-i kohandatud klastrite jaoks ja kasutasin kaheksat võimsat instantsi (C5.4xl) ning Sparklyr'i, et luua paindlikum tööprotsess…

# 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')
  )

…aga ülesanne ei olnud ikkagi täidetud. Mulle meeldisigaseda mitu moodi: suurendasin mälu iga päringu täitja jaoks, kasutasin rohkem mälu omavaid node'e, rakendasin laialehtseid muutujaid (broadcasting variable), kuid iga kord olid need ajutised lahendused ja lõpuks hakkasid täitjad kokku kukkuma, kuni kõik seiskus.

Uuendus: nii see algab. pic.twitter.com/agY4GU2ru5

— Nick Strayer (@NicholasStrayer) 15. mai 2019

Olen leidlikum

Mida ma õppisin: mõnikord vajavad erilised andmed erilisi lahendusi.

Iga SNP-l on positsiooni väärtus. See number vastab nende aluspaaride arvule, mis paikneb nende kromosoomis. See on hea ja loogiline viis meie andmete organiseerimiseks. Alguses tahtsin jagada kromosoomide alade kaupa. Näiteks positsioonid 1 — 2000, 2001 — 4000 jne. Kuid probleem on selles, et SNP-d on kromosoomidel jaotatud ebavõrdselt, seega rühmade suurused oleksid väga erinevad.

Seetõttu jõudsin ma positsioonide kategooriate (rank) jagamiseni. Juba laaditud andmete põhjal jooksutasin päringu unikaalsete SNP-de, nende positsioonide ja kromosoomide loetelu saamiseks. Seejärel sorteerisin andmed iga kromosoomi sees ja koondasin SNP-d määratud suurusega rühmadesse (bin). Ütleme, et iga rühm sisaldas 1000 SNP-d. See andis mulle seose SNP-de ja kromosoope rühma vahel.

Lõpuks tegin rühmad (bin) 75 SNP-st, põhjuse selgitan allpool.

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

Esimene katse Sparkiga

Mida ma õppisin: Sparkis ühendamine töötab kiiresti, kuid jaotamine on endiselt kulukas.

Tahas lugeda seda väikest (2,5 miljonit rida) andmevahemikku Sparkis, ühendada see toorandmetega ja seejärel jagada värskelt lisatud veeru järgi. 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')
  )

Ma kasutasin sdf_broadcast(), nii et Spark teab, et ta peab andmevahemiku kõikidele sõlmedele saatma. See on kasulik, kui andmed on väikesemõõtmelised ja kuna neid vajatakse kõigis ülesannetes. Vastasel juhul püüab Spark olla nutikas ja jaotab andmeid vastavalt vajadusele, mis võib põhjustada viivitusi.

Ja jälle ei töötanud mu idee: ülesanded töötasid mingisugust aega, lõpetasid ühendamise, kuid seejärel, nagu ka jaotamise täitjad, hakkasid nad ebaõnnestuma.

Lisa AWK

Mida ma õppisin: ärge magage, kui teile õpetatakse aluseid. Kindlasti on keegi teie probleemi juba 1980ndatel lahendanud.

Siiani on kõik mu ebaõnnestumised Sparkiga olnud põhjustatud andmete segadusest klastris. Võib-olla saaks olukorda parandada eeltöötlemisega. Otsustasin proovida jagada toorandmed kromosoomi veergudeks, lootes anda Sparkile "eelnevalt jaotatud" andmed.

Otsisin StackOverflow's, kuidas jagada veeru väärtuste järgi, ja leidsin nii toreda vastuse. AWK abil võite teksti faili jagada veeru väärtuste järgi, sooritades skriptis kirje, mitte saates tulemusi stdout.

Katsetamiseks kirjutasin Bash-skripti. Allalaadisin ühe pakendatud TSV faili, seejärel dekomprimeerisin selle abil gzip ja saatsin selle awk.

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

See töötas!

Protsessorite täitmine

Mida ma õppisin: gnu parallel — see on maagiline asi, kõik peaksid seda kasutama.

Jagamine kulges üsna aeglaselt, ja kui ma käivitasin htop, et kontrollida võimsate (ja kallite) EC2 instantside kasutamist, selgus, et kasutan ainult ühte tuuma ja umbes 200 MB mälu. Probleemi lahendamiseks ja mitte palju raha kaotamiseks pidi olema mõni viis töö paralleelseks muutmiseks. Õnneks leidsin ma täiesti hämmastavast raamatust Data Science at the Command Line Jeroen Jansseni, peatüki, mis käsitleb paralleelsust. Seal õppisin ma gnu parallel, väga paindlikku meetodit mitme lõime rakendamiseks Unixis.

Kui ma alustasin jagamist uue protsessiga, oli kõik hästi, kuid kitsaskohaks jäi S3 objektide alla laadimine kettale, mis ei olnud kuigi kiire ega täielikult paralleelne. Selle parandamiseks tegin järgmist:

1. Selgitasin välja, et S3 allalaadimise etappi saab otse toru sisse realiseerida, täielikult välistades vahepealse salvestamise ketas. See tähendab, et ma võin vältida toorandmete kirjutamist kettale ja kasutada veelgi väiksemat ja seega odavamat salvestusruumi AWS-is.
2. Käsk aws configure set default.s3.max_concurrent_requests 50 suurendas oluliselt AWS CLI kasutatavate lõimede arvu (vaikesätte järgi 10).
3. Läksin üle kiirematele EC2 instantsidele, mille nimetas on 'n'. Tulin välja, et n-instantidega arvutusvõimsuse kaotus kompenseeritakse suurema allalaadimise kiirusena. Enamikus ülesannetes kasutasin c5n.4xl.
4. Muutsin gzip järgnevaga pigz, see on gzip tööriist, mis oskab teha lahedaid asju algselt mitte paralleelse failide dekompressimise ülesande paralleelseks muutmiseks (see aitas kõige vähem).

# 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

Need sammud on omavahel kombineeritud, et kõik töötaks väga kiiresti. Allalaadimise kiiruse suurenemise ja kettale kirjutamise loobumise tõttu suutsin nüüd töödelda 5 terabaidi paketti vaid mõne tunni jooksul.

Pole midagi magusamat, kui näha, et kõik tuumad, mille eest sa AWS-is maksad, on kasutusel. Tänu gnu-parallelile saan ma zipida ja jagada 19 GB CSV faili sama kiiresti kui ma selle alla laadin. Ma ei suutnud isegi Spark'i selleks käivitada. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) 17. mai 2019

Selles tweetis oleks pidanud olema 'TSV'. Kahjuks ei olnud.

Taaskordatud andmete kasutamine

Mida ma õppisin: Spark eelistab kokkusurumata andmeid ja ei soovi partitsioone kombineerida.

Nüüd asusid andmed S3-s pakendamata (loe: jagatud) ja poolkorraldatud formaadis, ning ma sain jälle tagasi minna Spark'i juurde. Mind ootas üllatus: ma ei saanud taas soovitud tulemusi! Oli väga keeruline öelda Spark'ile, kuidas andmed on partitsioneeritud. Ja isegi kui ma seda tegin, selgus, et partitsioone oli liiga palju (95 tuhat), ja kui ma kasutasin coalesce nende arvu vähendamiseks mõistlikesse piiridesse, rikkus see mu partitsioneerimise. Olen kindel, et seda on võimalik parandada, kuid paaripäevaste otsingute jooksul ei suutnud ma lahendust leida. Lõpuks lõpetasin kõik ülesanded Spark'is, kuigi see võttis aega, ja mu jagatud Parquet-failid ei olnud just väikesed (~200 Kb). Kuid andmed asusid seal, kus nad pidid olema.

Liiga väikesed ja erinevad, imeline!

Kohalike Spark-päringute testimine

Mida ma õppisin: Spark'is on liiga palju kulusid lihtsate ülesannete lahendamisel.

Laadides andmed läbi mõeldud vormingu, suutsin testida kiirus. Seadistasin R-i skripti kohaliku Spark-serveri käivitamiseks ja seejärel laadisin Spark'i andmeframe'e määratud Parquet-gruppidest (bin). Üritasin laadida kõiki andmeid, kuid ei suutnud Sparklyr'i veenduda partitsioneerimise tuvastamises.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Alusta taimerit
start_time <- Sys.time()

# Laadi soovitud bin Spark'i
intensity_data % 
  Spark_read_Parquet(
    name = 'intensity_data', 
    path = get_snp_location(desired_snp),
    memory = FALSE )

# Filtreeri bin snp-ks ja seejärel kogu kohalikku
test_subset % 
  filter(SNP_Name == desired_snp) %>% 
  collect()

print(Sys.time() - start_time)

Sooritamine võttis 29,415 sekundit. Tõhusam, kuid mitte eriti hea massiliste testide jaoks. Lisaks ei suutnud ma kiirusel töötamist kiirendada mälukaitse abil, kuna kui üritasin andmeframe'e mällu vahemällu salvestada, kukkus Spark alati kokku, isegi kui ma eraldasin rohkem kui 50 GB mälu andmestikule, mis kaalub vähem kui 15.

Tagasi AWK juurde

Mida ma õppisin: assotsiatiivsed massiivid AWK-s on väga tõhusad.

Mõistsin, et saan saavutada suuremat kiirus. Meeles tuli, et ma lugesin suurepärasest Bruce Barnetti AWK käsiraamatust üht lahedat funktsiooni, mis nimetatakse "assotsiatiivsed massiivid". Tegelikult on need võtme-väärtuse paarid, mida erinevatel põhjustel on AWK-s nimetatud teistmoodi, mistõttu ma ei tulnud neid eriti meelde. Roman Cheplyaka tuletas meelde, et termin "assotsiatiivsed massiivid" on palju vanem kui termin "võtme-väärtuse paar". Isegi kui otsite võtme-väärtuse paari Google Ngram's, ei leia te seda terminit, kuid assotsiatiivsed massiivid on seal olemas! Samuti tõlgendatakse "võtme-väärtuse paar" kõikjal andmebaasidega, seega on loogilisem võrrelda hashmap'iga. Mõistsin, et saan kasutada neid assotsiatiivseid massiive, et siduda mu SNP-d bin-tabeli ja toorandmetega ilma Spark'ita. Selleks kasutasin AWK skriptis blokki, seda terminit te seal ei näe, kuid leiate assotsiatiivsed massiivid! Samuti seondub „võti-väärtus“ kõige sagedamini andmebaasidega, seega on palju loogilisem võrrelda hashmapiga. Ma sain aru, et saan neid assotsiatiivseid massiive kasutada oma SNP-de sidumiseks grupitabeliga (bin table) ja toorandmetega ilma Spark’i kasutamiseta.

Selleks kasutasin AWK-skripti plokki BEGIN. See on koodifragmendi, mis täidetakse enne, kui esimene andmereale saadetakse skripti põhiosasse.

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

Meeskond while(getline...) laadis kõik read gruppide (bin) CSV-st, määrates esimese veeru (SNP nime) assotsiatiivse massiivi võtmena bin ja teise väärtuse (grupi) väärtusena. Seejärel rakendati põhifaili kõigi read sisaldavas blokis, milles iga rida saadetakse väljundfaili, mis sai unikaalse nime sõltuvalt oma grupist (bin): { }..._bin_"bin[$1]"_... ..._bin_"bin[$1]"_....

Muutujad batch_num ja chunk_id vastasid konveieri esitatud andmetele, mis aitas vältida võistlustingimusi, ja iga käivitamisel käivitatud paralleel, kirjutas oma unikaalsesse faili.

Kuna kõik toorandmed jaotasin kaustades kromosoomide järgi, mis jäid üle minu varasemast katsetest AWK-ga, sain nüüd kirjutada teise Bash skripti, et töödelda kromosoomi kaupa ja edastada S3-sse sügavamalt partitsioneeritud andmed.

DESIRED_CHR='13'

# Laadi kromosoomi andmed s3-st ja jaga gruppideks
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"

# Ühenda kõik paralleelsed protsesside killud üheks failiks ja laadi rds-ks üles R-i abil
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/*

Skripti kaks osa paralleel.

Esimeses jaotises loetakse andmed kõikidest failidest, mis sisaldavad vajalikku kromosoomi teavet, seejärel jaotatakse need andmed voogudesse, mis jagavad failid vastavatesse gruppidesse (bin). Et vältida olukordi, kus mitu voogu kirjutavad ühte faili, edastab AWK failide nimed, kuhu andmeid kirjutada, näiteks, chr_10_bin_52_batch_2_aa.csv. Tulemuseks luuakse kettale palju väikeseid faile (selleks kasutasin terabaidiseid EBS-ükseid).

Teise jaotise toru paralleel läbib gruppe (bin) ja ühendab nende eraldi failid üldiseks CSV-ks, cat, ja seejärel saadetakse need ekspordiks.

Ülekandmine R-i?

Mida ma õppisin: sellele saab juurde pääseda stdin ja stdout R-skripti kaudu, mis tähendab, et seda saab kasutada ka torus.

Bash-skripti sees võisite näha sellist rida: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... See edastab kõik ühendatud grupi (bin) failid allolevale R-skripti. {} on spetsiaalne tehnika paralleel, mis kõik edastatud andmed määratud voogu sisestab otse käsusse. Valik {#} annab unikaalse ID täitmisvoogule ja {%} esindab ülesande sloti numbrit (see kordub, kuid mitte kunagi samal ajal). Kõiki valikute loendit saab leida dokumentatsioonis.

#!/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
)

Kui muutuja file("stdin") edastatakse readr::read_csv, andmed, mis on edastatud R-skripti, laaditakse raamisse, mis seejärel salvestatakse .rds-failina, kasutades aws.s3 otse S3-sse.

RDS on nagu Parquet'i noorem versioon, ilma veeruhoiustamise keerukusteta.

Pärast Bash-skripti lõpetamist sain hunniku .rds-faile, mis asusid S3-s, mis võimaldas mul kasutada tõhusat tihendamist ja sisseehitatud tüüpe.

Hoolimata aeglasest R-ist töötas kõik väga kiiresti. Pole üllatav, et R-is andmete lugemist ja kirjutamist käsitlevad fragmendid on hästi optimeeritud. Keskmise suurusega kromosoomi testimisel viidi ülesanne C5n.4xl instantsil lõpule umbes kahe tunni jooksul.

S3 piirangud

Mida ma õppisin: nutika S3 tee rakendamise tõttu suudab see töödelda palju faile.

Olin mures, kas S3 suudab töödelda paljusid sellele edastatud faile. Ma oleksin võinud teha failide nimed arusaadavaks, kuid kuidas S3 neid leiab?

Kataloogid S3-s on vaid ilu nimel, tegelikult ei huvitavad süsteemi sümbolid /. S3 KKK-lehelt.

Tundub, et S3 esitab tee konkreetse faili juurde kui lihtsa võtme ebatüüpilises hunniku tabelis või dokumentide andmebaasis. Bucketit (konteinerit) võib pidada tabeliks ja faile — selle tabeli rekorditeks.

Kuna kiirus ja efektiivsus on Amazonis kasumi teenimise jaoks olulised, ei ole üllatav, et see "võti-failiteele" süsteem on erakordselt optimeeritud. Üritasin leida tasakaalu: et ei peaks tegema palju get-päringuid, kuid et päringud toimuksid kiiresti. Selgus, et parim on teha umbes 20 000 bin-faili. Arvan, et kui jätkata optimeerimist, võib saavutada väiksemat otsingulauda (näiteks luues spetsiaalse konteineri ainult andmete jaoks). Kuid edasiste katsetuste jaoks ei olnud enam aega ja raha.

Kuidas on ristühilduvusega?

Mida ma õppisin: peamine aja raiskamise põhjus on teie failide salvestamise meetodi enneaegne optimeerimine.

Sel hetkel on väga oluline eneselt küsida: „Miks kasutada patenteeritud failiformaate?” Põhjus peitub laadimiskiirusest (gzip pakitud CSV-failide laadimine kestis 7 korda kauem) ja meie töövoogude ühilduvusest. Ma võin oma otsust ümber vaadata, kui R suudab hõlpsasti laadida Parquet-failide (või Arrow) formaate ilma Spark'i koormuseta. Meie laboris kasutavad kõik R-i ja kui mul on vaja andmed mõnda teise formaati konverteerida, on mul endiselt algsed tekstifailid, nii et ma võin lihtsalt toru uuesti käivitada.

Töö jagamine

Mida ma õppisin: ärge proovige ülesandeid käsitsi optimeerida, las see teeb arvuti.

Ma testisin töövoogu ühe kromosoomiga, nüüd on vaja töödelda kõik ülejäänud andmed.
Soovisin tõsta mitu EC2 instantsi konverteerimiseks, kuid samas olin mures, et saan äärmiselt tasakaalustamata koormuse erinevates töötlemise ülesannetes (nagu ka Spark kannatas tasakaalustamata partitsioonide all). Lisaks ei olnud ma valmis tõstma ühte instantsi iga kromosoomi jaoks, sest AWS-i kontodel on vaikimisi piirang 10 instantsi.

Siis otsustasin kirjutada R-skript ülesannete töötlemise optimeerimiseks.

Küsin esmalt S3-lt, kui palju ruumi iga kromosoomi salvestus võtab.

library(aws.s3)
library(tidyverse)

chr_sizes % 
  mutate(Size = as.numeric(Size)) %>% 
  filter(Size != 0) %>% 
  mutate(
    # Tõmba kromosoom faili nimest 
    chr = str_extract(Key, 'chr.{1,4}.csv') %>%
             str_remove_all('chr|.csv')
  ) %>% 
  group_by(chr) %>% 
  summarise(total_size = sum(Size)/1e+9) # Jagamine, et saada väärtus GB-des



# Tibble: 27 x 2
   chr   total_size
         
 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.
# … veel 17 rida

Seejärel kirjutasin funktsiooni, mis võtab kokku suuruse, segab kromosoomide järjekorda ja jagab need gruppidesse num_jobs ja teatab, kui palju erinevad kõik töötlemise ülesannete suurused.

num_jobs <- 7
# Kui suur oleks iga töö, kui need oleksid ideaalselt jaotatud?
job_size <- sum(chr_sizes$total_size)/7

shuffle_job %
    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)



# Tibble: 1 x 2
     sd data            
             
1  153.

Siis jooksin purrriga tuhat segamist ja valisin parima.

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

Nii sain ma ülesannete komplekti, mis oli suuruse poolest üsna ühtlane. Seejärel jäi mul vaid mu varasem Bash-skript suuresse tsüklisse panna. for. Selle optimeerimise kirjutamiseks kulus umbes 10 minutit. Ja see on tunduvalt vähem, kui ma raiskaksin tasakaalustamatuse korral ülesannete käsitsi loomisele. Seega arvan, et see eelnevalt tehtud optimeerimine oli kasulik.

for DESIRED_CHR in "16" "9" "7" "21" "MT"
do
# Kood ühe kromosoomi töötlemiseks
fi

Lõpuks lisan välja lülitamise käsu:

sudo shutdown -h now

… ja kõik töötas! AWS CLI abil käivitasin instantsid ja optsiooniga user_data edas kandis neile nende töötlemise ülesannete Bash-skriptid. Need täideti ja lülitati automaatselt välja, nii et ma ei maksnud liigsest arvutusvõimsusest.

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

Pakime kokku!

Mida ma õppisin: API peab olema lihtne lihtsuse ja paindlikkuse pärast.

Lõpuks sain andmed õiges kohas ja vormingus. Jäi veel maksimaalselt lihtsustada andmete kasutamise protsessi, et mu kolleegidel oleks kergem. Tahtsin luua lihtsa API päringute koostamiseks. Kui ma tulevikus otsustan liikuda .rds Parquet-failidele, siis peab see olema probleem minu, mitte kolleegide jaoks. Selleks otsustasin luua sisemise R-paketi.

Kogusin ja dokumenteerisin väga lihtsa paketi, mis sisaldab vaid mõnda funktsiooni andmete ligipääsuks, keskendudes funktsioonile get_snp. Samuti tegin kolleegidele saidi pkgdown, et nad saaksid lihtsalt näha näiteid ja dokumentatsiooni.

Intelligentne vahemälu

Mida ma õppisin: kui teie andmed on hästi ette valmistatud, siis on vahemälule lihtne!

Kuna üks peamisi töövooge rakendas sama analüüsimudeli SNP paketile, otsustasin kasutada rühmitamist (binning) enda kasuks. SNP-de andmeid edastades lisatakse naasvale objektile kogu rühma (bin) teave. See tähendab, et vanad päringud võivad (teoreetiliselt) kiirendada uute päringute töötlemist.

# 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
  }
...

Paketti koostades jooksin ma läbi palju võrdlusi, et võrrelda erinevate meetodite kiirus. Soovitan seda mitte alahinnata, kuna mõnikord on tulemused ootamatud. Näiteks dplyr::filter olid tunduvalt kiiremad kui reastamise järgi filtrimise baasil, samas kui ühe veeru saamine filtreeritud andmeframist töötas palju kiiremini kui indekseerimise süntaks.

Pange tähele, et objekt prev_snp_results sisaldab võtmeid snps_in_bin. See on kõigi unikaalsete SNP-de massiiv grupis (bin), mis võimaldab kiiresti kontrollida, kas eelmisest päringust on juba andmed. Samuti lihtsustab see rühma (bin) kõigi SNP-de iteratiivset läbisõitu järgmise koodiga:

# 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 
}

Tulemused

Nüüd saame (ja hakkasime tõsiselt) jooksma mudeleid ja stsenaariume, mis varem olid meile kättesaamatud. Parim on see, et mu kolleegid laborist ei pea muretsema igasuguste keerukuste üle. Neil on lihtsalt toimiv funktsioon.

Ja kuigi paketil on neid üksikasju, püüdsin andmeformaadi teha piisavalt lihtsaks, et nad saaksid sellest aru, kui ma homme äkki kaon…

Kiirus on märkimisväärselt suurenenud. Tüüpiliselt skaneerime funktsionaalselt olulisi genome osi. Varem ei saanud me seda teha (oli liiga kallis), kuid nüüd grupi (bin) struktuuri ja vahemälu abil võtab ühe SNP päring ainult umbes 0,1 sekundit, ja andmete kasutamine on nii madal, et S3 kulud on tühised.

Hiljuti määrati mind korraldama 25+ TB toore genotüpiseerimise andmeid oma laboratooriumis. Kui ma alustasin, võttis Spark kasutamine 8 minutit ja maksis 20 dollarit SNP pärimiseks. Pärast AWK + #rstats töötlemiseks kulub nüüd vähem kui kümnendik sekundist ja maksab $0.00001. Minu isiklik #BigData võit. pic.twitter.com/ANOXVGrmkk

— Nick Strayer (@NicholasStrayer) 30. mai 2019

Kokkuvõte

See artikkel ei ole juhend. Lahendus osutus individuaalseks ja peaaegu kindlasti mitte optimaalseks. Pigem on see jutustus teekonnast. Soovin, et teised mõistaksid, et sellised lahendused ei tule pähe täielikult kujundatuna, vaid on proovimise ja eksimise tulemus. Lisaks, kui otsite andmeanalüüsi spetsialisti, siis pidage meeles, et nende tööriistade tõhusaks kasutamiseks on vajalik kogemus, mis vajab raha. Olen õnnelik, et mul oli ressursse maksmiseks, kuid paljude teiste jaoks, kes võivad teha sama tööd paremini, ei pruugi kunagi olla sellist võimalust raha puudumise tõttu isegi katsetamiseks.

Suured andmed on universaalsed tööriistad. Kui teil on aega, siis tõenäoliselt suudate kirjutada kiirem lahendus, kasutades 'nutikat' andmete puhastamist, ladustamist ja ekstraktsiooni meetodeid. Lõppkokkuvõttes seisneb kõik kulude ja kasu analüüsis.

Mida olen õppinud:

• ei ole odavat viisi 25 TB korraga käsitleda;
• olge ettevaatlik oma Parquet-failide suuruse ja korraldusega;
• Sparkis peavad partisioonid olema tasakaalus;
• ärge kunagi püüdke teha 2,5 miljonit partisiooni;
• sorteerimine on endiselt keeruline, nagu ka Spark'i seadistamine;
• mõnikord vajavad erilised andmed erilisi lahendusi;
• Sparkis toimivad kokkuvõtted kiiresti, kuid partisioneerimine on endiselt kulukas;
• ärge magage, kui teile õpetatakse aluseid, keegi on tõenäoliselt teie probleemi juba 1980ndatel lahendanud;
• gnu parallel — see on maagiline asi, kõik peaksid seda kasutama;
• Spark armastab kokku surmatud andmeid ja ei armasta partisioonide ühendamist;
• Sparkis on liiga palju kulusid, kui lahendatakse lihtsaid ülesandeid;
• assotsiatiivsed massiivid AWK-s on väga tõhusad;
• saate neid kutsuda stdin ja stdout R-skriptist, seega ka kasutada neid voos;
• nutika S3 teostuse abil suudab töödelda palju faile;
• peamine ajakulu põhjus on teie ladustamismeetodi enneaegne optimeerimine;
• ärge proovige ülesandeid käsitsi optimeerida, laske arvutil seda teha;
• API peab olema lihtne lihtsuse ja paindlikkuse nimel;
• kui teie andmed on hästi ette valmistatud, on vahemäluga töötamine lihtne!

Allikas: habr.com