Kuidas seda artiklit lugeda: palun vabandust, et tekst läks nii pikaks ja segaseks. Ajan teie aega säästmiseks alustasin iga peatüki sissejuhatusega „Mida ma õppisin”, kus ma lühidalt selgitän peatüki sisu ühe või kahe lausega.

„Lihtsalt näita lahendust!” Kui soovite lihtsalt näha, kuhu ma jõudsin, minge peatükki „Ajan end leidlikumaks”, kuid ma arvan, et huvitavam ja kasulikum on lugeda ebaõnnestumistest.

Hiljuti anti mulle ülesanne seadistada protsess suure koguse DNA algse järjestuse (tehniliselt SNP-chip) töötlemiseks. Selleks pidi kiiresti saama teavet antud geneetilise asukoha (mida kutsutakse SNP-ks) kohta, et hiljem saaks tegeleda modelleerimise ja muude ülesannetega. Kasutades R ja AWK, sain andmeid loogiliselt puhastada ja korraldada, mis kiirendas oluliselt päringute töötlemist. See ei olnud lihtne ja nõudis arvukaid iteratsioone. See artikkel aitab teil vältida mõningaid minu vigu ja demonstreerib, millega ma lõpuks hakkama sain.

Alustuseks mõned tutvustavad selgitused.

Andmed

Meie ülikooli geneetilise informatsiooni töötlemise keskus andis meile andmeid TSV vormingus, kogus 25 TB. Need olid jagatud 5 paketti, Gzip-i tihendatud, kus igaühes oli umbes 240 nelja gigabaidi faili. Iga rida sisaldas andmeid ühe SNP kohta ühe isiku kohta. Kokku edastati andmed umbes 2,5 miljoni SNP ja umbes 60 tuhande inimese kohta. Peale SNP-teabe sisaldasid failid palju veerge numbritega, mis kajastasid erinevaid omadusi, nagu lugemise intensiivsus, erinevate alleelide sagedus jne. Kokku oli umbes 30 veergu unikaalsete väärtustega.

Eesmärk

Nagu igas andmehälgema projektis, oli kõige olulisem määrata, kuidas andmeid kasutatakse. Antud juhul peame enamasti valima mudeleid ja töövooge SNP-de põhjal SNP-de järgi. See tähendab, et samal ajal on meil vaja andmeid ainult ühe SNP kohta. Ma pidin õppima, kuidas kõige lihtsamalt, kiiremini ja odavamalt välja võtta kõik kirjed, mis on seotud ühe 2,5 miljoni SNP-st.

Kuidas seda mitte teha

Tsiteerin sobivat klišeed:

Ma ei ole tuhat korda ebaõnnestunud, vaid avastasin tuhat viisi, kuidas mitte andmeid hõlpsaks päringuks parseerida.

Esimene katse

Mida ma õppisin: ei ole odavat viisi 25 TB korraga eemaldada.

Kuulates Vanderbilti Ülikooli kursust "Suured andmed: arenenud töötlemismeetodid", olin kindel, et see on lihtne. Küllap kulub tund või kaks Hive-serveri seadistamiseks, et kõik andmed läbi käia ja tulemusi esitada. Kuna meie andmed asuvad AWS S3-s, kasutasin teenust Athena, mis võimaldab rakendada Hive SQL-päringuid S3 andmetele. Hive-kluster ei vaja seadistamist/ülesehitamist ja maksad vaid nende andmete eest, mida otsid.

Pärast seda, kui näitasin Athenale oma andmeid ja nende vormingut, jooksin mõned testid sarnaste päringutega:

select * from intensityData limit 10;

Ja sain kiiresti korralikult struktureeritud tulemusi. Valmis.

Kuni pole proovinud andmeid töös…

Paluti mul välja võtta kogu info SNP-de kohta, et testida oma mudelit. Käivitasin päringu:

select * from intensityData 
where snp = 'rs123456';

…ja hakkasin ootama. Kaheksa minuti ja üle 4 TB küsitud andmete järel sain tulemuse. Athena tasub leitud andmete mahu eest, $5 terabaidi kohta. Seega maksis see ainus päring $20 ja ooteaega kaheksa minutit. Mudeli läbiviimiseks kõigi andmete põhjal oleks pidanud ootama 38 aastat ja maksma $50 miljonit. Ilmselgelt ei sobinud see meile.

Pidiime kasutama Parquet…

Mida ma õppisin: olge ettevaatlikud oma Parquet-failide suuruse ja korraldusega.

Esialgu proovisin olukorda parandada, konverteerides kõik TSV-d Parquet-failideks. Need on mugavad suure andmehulga töötlemiseks, sest teave on salvestatud veergude kaupa: iga veerg asub enda mälusegmendis/diskis, erinevalt tekstifailidest, kus read sisaldavad iga veeru elemente. Ja kui midagi otsida, piisab vajalikust veerust lugemisest. Lisaks sisaldab iga faili veerus väärtuste vahemik, nii et kui otsitav väärtus ei ole veeru vahemikus, ei raiska Spark aega kogu faili skaneerimisele.

Käivitasin lihtsa ülesande AWS Glue kuidas meie TSV-d Parquet-ks konverteerida ja uued failid Athenasse üles laadida. Selleks kulus umbes 5 tundi. Kuid kui ma käivitasin päringu, siis selle täitmiseks kulus umbes sama palju aega ja veidi vähem raha. Asi on selles, et Spark, proovides ülesannet optimeerida, lihtsalt pakkis ühe TSV-osa lahti ja pani selle oma Parquet-osa. Ja kuna iga osa oli piisavalt suur ja sisaldas tervet rida paljude inimeste registreeringutest, siis sisaldas iga fail kõiki SNP-sid, seetõttu pidi Spark avama kõik failid, et otsida vajalikku teavet.

Huvitav on see, et Parquet'i vaikimisi (ja soovitatav) kompressioonitüüp — snappy — ei ole jagatav (splitable). Seetõttu takerdus iga täitja (executor) ülesande lahti pakkimise ja kogu 3,5 GB suuruse andmesessiooni üleslaadimise peale.

Käime probleemiga läbi

Mida ma õppisin: sortimine on keeruline, eriti kui andmed on hajutatud.

Mul näis, et nüüd sain probleemist aru. Peasin lihtsalt andmed sorteerima SNP veeru alusel, mitte inimeste alusel. Siis säilitatakse eraldi andmeosas mitu SNP-d ja siis tuleb pargu „nutikas“ funktsioon mängu, mis ütleb „ava ainult siis, kui väärtus on vahemikus“. Kahjuks osutus miljardeid ridu, mis olid hajutatud klastrisse, sorteerimine keerukaks ülesandeks.

Mina võtmas algoritmide kursust kolledžis: «Kehv, kedagi ei huvita nende kõigi sortimisalgoritmide arvutustüübid»

Mina üritamas sortida 20TB-s veerus #spark tabel: «Miks see nii kaua aega võtab?» #DataScience vaevad.

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

AWS ei taha kindlasti raha tagasi maksta põhjendusega „Ma olen hajevil üliõpilane“. Pärast seda, kui käivitasin sortimise Amazon Glue'is, töötas see 2 päeva ja lõpetas ebaõnnestumisega.

Kuidas on partitionimist?

Mida ma õppisin: Sparkis peavad osad olema tasakaalustatud.

Siis tuli mulle idee andmeid kromosoomide järgi partitionsida. Neid on 23 (ja veel mõned, kui arvestada mitoohondria DNA-d ja dekodeerimata alasid).
See võimaldab andmed jagada väiksemateks osadeks. Kui lisada Spark-funktsiooni ekspordi skripti Glue'ile lihtsalt üks rida partition_by = "chr", siis andmed peaksid olema jagatud kärudesse (buckets).

Genoom koosneb arvukatest fragmentidest, mida nimetatakse kromosoomideks.

Kahjuks see ei toiminud. Chromosoomidel on erinevad suurused, mis tähendab ka erinevat teabe hulka. See tähendab, et Spark'i kaudu töötajatele saadetud ülesanded ei olnud tasakaalus ja täideti aeglaselt, kuna mõned sõlmed lõpetasid varem ja seisisid. Kuid ülesanded viidi ellu. Siiski, kui küsiti üht SNP, taasiseseisvus tasakaalustamatus probleemideks. SNP töötlemise maksumus suuremates kromosoomides (st sealt, kust me andmeid soovime saada) vähenes vaid umbes 10 korda. Palju, aga mitte piisavalt.

Ent kui jagada veel väiksemateks partiideks?

Mida ma õppisin: üldiselt ärge proovige kunagi teha 2,5 miljonit partiid.

Otsustasin kõike proovida ja partitseerisin iga SNP. See tagas, et partii suurused oleksid ühtlased. SEE OLI HALB IDEE. Kasutasin Glue'i ja lisasin süütu rea partition_by = 'snp'. Ülesanne käivitati ja hakkas jooksma. Päeva pärast kontrollisin ja nägin, et S3-s pole ikka veel midagi salvestatud, nii et lõpetasin ülesande. Tundub, et Glue salvestas vahefailid S3-sse varjatud kohta, ja palju faile, võib-olla paar miljonit. Selle tulemusena maksis minu viga rohkem kui tuhat dollarit ja ei rõõmustanud mu mentorit.

Partitsioneerimine + sortimine

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

Viimane katse partitsioneerimise osas oli see, et partitsioneerisin kromosoome ja seejärel sortisin iga partii. Teoreetiliselt peaks see kiirendama iga päringut, kuna soovitud SNP andmed peaksid olema mitme Parquet'i ploki piires antud vahemikus. Kahjuks osutus isegi partitsioneeritud andmete sortimine keeruliseks ülesandeks. Tulemuseks läksin üle EMR-ile kohandatud klastriga 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')
  )

…kuid ülesanne ei olnud siiski lõpuks täidetud. Seadsin asja igat pidi: suurendasin mälu eraldamist iga päringu teostaja jaoks, kasutasin suure mälu mahuga sõlmi, rakendasin laialehe muutujaid (broadcasting variable), kuid iga kord oli see osaline lahendus ja järk-järgult hakkasid teostajad ebaõnnestuma, kuni kõik seisis.

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

— Nick Strayer (@NicholasStrayer) 15. mai 2019

Muudan end leidlikumaks

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

Iga SNP-l on positsiooniväärtus. See on number, mis vastab aluste arvule, mis asuvad selle kromosoomi kõrval. See on hea ja loomulik viis meie andmete korraldamiseks. Alguses tahtsin rühmitada iga kromosoomi piirkondade järgi. Näiteks positsioonid 1–2000, 2001–4000 jne. Kuid probleem on selles, et SNP-d on kromosoomides ebaühtlaselt jaotatud, seega on grupi suurused oluliselt erinevad.

Tulemusena jõudsin ma positsioonide kategooriate (rank) jagamiseni. Laaditud andmete põhjal jooksin päringu, et saada unikaalsete SNP-de, nende positsioonide ja kromosoomide nimekiri. Seejärel sorteerisin andmed iga kromosoomi sees ja kogusin SNP-d antud suurusega gruppidesse (bin). Ütleme, 1000 SNP kohta. See andis mulle SNP-de seose grupiga kromosoomis.

Lõpuks tegin rühmadesse (bin) 75 SNP, 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: ühendamine Sparkis toimub kiiresti, kuid partitsioneerimine on endiselt kulukas.

Tahtsin laadida selle väikese (2,5 miljonit rida) andmeframes Sparkisse, ühendada selle toorandmetega ja seejärel partitsioneerida uue 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')
  )

Olen kasutanud sdf_broadcast(), nii teab Spark, et ta peab andmeframe kõikidesse sõlmepunktidesse saatma. See on kasulik, kui andmed on väikesed ja vajalikud kõigi ülesannete jaoks. 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 plaan: ülesanded töötasid mingil ajal, lõpetasid ühendamise, ja siis, nagu ka käivitatud partitsioneerimise täitjad, hakkasid need ebaõnnestuma.

Lisan AWK

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

Kuni selle hetkeni oli kõigi minu ebaõnnestumiste põhjus Sparkis andmete segamine klastris. Võib-olla on olukorda võimalik parandada eelneva töötlemisega. Otsustasin proovida jagada toorandmed kromosoomi veergudeks, nii lootsin pakkuda Sparkile 'eelnevalt partitsioneeritud' andmeid.

Otsisin StackOverflow'ist, kuidas jagada veergude väärtuste järgi, ja leidsin nii imelise vastuse. AWK abil on võimalik jagada tekstifaili veergude väärtuste järgi, kirjutades skripti ja mitte suunates tulemusi stdout.

Katsena kirjutasin Bash-skripti. Laadisin alla ühe pakitud TSV faili ja seejärel dekompressisin selle gzip ja suunasin selle awk.

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

See töötas!

Tuuma täitmine

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

Jagamine toimus üsna aeglaselt ja kui ma käivitasin htop, et kontrollida võimsuse (ja kallite) EC2 instantsi kasutust, selgus, et kasutan ainult ühte tuuma ja umbes 200 MB mälu. Probleemi lahendamiseks, et mitte palju raha kaotada, pidi olema mõeldud, kuidas töö paralleelseks muuta. Õnneks leidsin ma täiesti hämmastavast raamatust Data Science at the Command Line Jeroen Janssensilt peatüki, mis oli pühendatud paralleele töötlemisele. Sealt sain teada gnu parallel, väga paindlik meetod mitme lõime rakendamiseks Unixis.

Kui käivitasin jagamise uue protsessi abil, oli kõik suurepärane, kuid jäi kitsaskohaks — S3 objektide allalaadimine kettale ei olnud liiga kiire ja ei olnud täielikult paralleelne. Selle parandamiseks tegin nii:

1. Selgitasin välja, et S3 allalaadimise etappi saab otse voos rakendada, täielikult välistades vahehoidla kettal. See tähendab, et saan vältida toorete andmete kirjutamist kettale ja kasutada veel väiksemat, seega ka odavamat ladustamist AWS-is.
2. Käsuga aws configure set default.s3.max_concurrent_requests 50 suurendas oluliselt lõime arvu, mida AWS CLI kasutab (vaikimisi on neid 10).
3. Läksin üle võrgu kaudu optimeeritud EC2 instantsile, mille nimetusel on täht n. Avastasin, et arvutusvõime kadu n-instantside kasutamisel kompenseerib suuresti allalaadimise kiirus.
4. Muutsin gzip . Tundub, et pigz, see on gzip tööriist, mis suudab teha suurepäraseid asju algselt paralleelse töötlemise rakendamiseks failide dekompressimisel (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 toimiks väga kiiresti. Kiire allalaadimise ja kettale kirjutamise loobumise tõttu sain nüüd töödelda 5 terabaidist paketti vaid paari tunni jooksul.

Mitte pole midagi magusamat kui näha, et kõik teie AWS-is tasutud tuumad on kasutuses. Tänu gnu-parallelile saan ma 19gig CSV faili sama kiiresti lahti pakkida ja jagada kui seda alla laadida. Ma ei suutnud isegi Spark'i tööle panna. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) 17. mai 2019

Selles tweets pidi olema mainitud 'TSV'. Kahjuks ei olnud.

Uute andmete kasutamine

Mida ma õppisin: Spark armastab tihendamata andmeid ja ei salli partitsioonide kombineerimist.

Nüüd olid andmed S3-s avatud (loe: jagatud) ja poolkorraldatud formaadis ning ma sain tagasi minna Spark'i juurde. Mind ootas üllatus: ma ei suutnud jälle soovitud tulemust saavutada! Oli väga keeruline öelda Spark'ile, kuidas andmed on partitsioneeritud. Ja isegi kui ma seda tegin, selgus, et partitsioone on liiga palju (95 tuhat), ja kui ma kasutasin coalesce et vähendada nende arvu mõistlikesse piiridesse, siis see rikkus minu partitsioneerimise. Olen kindel, et seda saab parandada, kuid paar päeva otsinguid ei toonud lahendust. Lõpuks tegin kõik ülesanded Spark'is ära, kuigi see võttis aega ja minu jagatud Parquet-failid ei olnud eriti väikesed (~200 Kb). Kuid andmed asusid seal, kus pidid.

Liialt väikesed ja erinevad, suurepärane!

Kohalike Spark'i päringute testimine

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

Andmete üleslaadimine hästi mõeldud formaadis võimaldas mul testida kiirus. Seadsin R-i skripti kohalikku Spark'i serverisse käivitamiseks ja seejärel laadisin Spark'i andmeframe'i määratud Parquet-gruppide (bin) salvestusest. Proovisem laadida kõik andmed, kuid ei suutnud Sparklyr'ile partitsioneerimist tuvastada.

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 )

# Alamhulk bin'ist snp ja seejärel kogu kohalikusse
test_subset % 
  filter(SNP_Name == desired_snp) %>% 
  collect()

print(Sys.time() - start_time)

Teostamine võttis 29,415 sekundit. Palju parem, kuid mitte piisavalt hea massilise testimise jaoks. Pealegi ei saanud ma kiirusetõusu saavutada mälus andmeframe'i vahemälustades, kuna iga kord, kui proovisin andmeframe'i mälu salvestada, kukkus Spark alati kokku, isegi kui eraldasin rohkem kui 50 GB mälu andmestikule, mis kaalus alla 15.

Tagasi AWK'i juurde

Mida ma õppisin: assotsiatiivsed massiivid AWK's on väga efektiivsed.

Mõistsin, et saan saavutada suuremat kiirus. Meenus, et suurepärases AWK juhendis Bruce Barnett'ilt Ma olin lugenud lahedast funktsioonist, mida nimetatakse „assotsiatiivsed massiivid”. Tegelikult on need võtme-väärtuse paarid, mida kuidagi AWK-s teisti nimetatakse, ja seetõttu ei tulnud ma nendele eriti mõelnud. Roman Cheplyaka tuletas meelde, et mõisted „assotsiatiivsed massiivid” on palju vanemad kui mõisted „võtme-väärtuse paar”. Isegi kui te otsite võtme-väärtuse paar Google Ngram-st, ei näe te seda mõistet seal, vaid leiate assotsiatiivsed massiivid! Pealegi seostatakse „võtme-väärtuse paare” enamasti andmebaasidega, seega on palju mõistlikum võrrelda neid hashmapiga. Ma sain aru, et saan neid assotsiatiivseid massiive kasutada oma SNP-de seostamiseks rühma tabeliga (bin table) ja toorandmete jaoks ilma Sparkita.

Selleks kasutasin AWK skriptis ploki BEGIN. See on koodijupp, mis täidetakse enne, kui esimene andmereegel edastatakse skripti peamisse keha.

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 CSV grupist (bin), määrates esimese veeru (SNP nimi) võtmena assotsiatiivses massiivis bin ja teise väärtuse (grupp) väärtusena. Seejärel plokis { }, mis täidetakse kõigi peafaile ridade kohta, saadetakse iga rida väljastusfaili, millel on ainulaadne nimi, sõltuvalt tema grupist (bin): ..._bin_"bin[$1]"_....

Muutujad batch_num ja chunk_id vastasid torujuhtme poolt esitatud andmetele, mis võimaldas vältida võidujooksu olukorda, ja igal käivitatud parallel, kirjutas oma ainulaadsesse faili.

Kuna kõik toorandmed olin jaganud kaustadesse kromosoomide järgi, mis olid jäänud pärast minu eelmist AWK eksperimenti, sain nüüd kirjutada teise Bash skripti, et töödelda kromosoomi korraga ja saata S3-sse sügavamalt partsioneeritud andmeid.

DESIRED_CHR='13'

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

# Koonda kõik paralleelsed protsessi osad ühte faili ja laadi RDS-sse, kasutades R-i
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/*

Skriptis on kaks osa parallel.

Esimese jaotises loetakse andmed kõigist failidest, mis sisaldavad teavet vajaliku kromosoomi kohta, seejärel jaotatakse need andmed voogude vahel, mis jaotavad failid vastavatesse gruppidesse (bin). Selleks, et vältida seaduse rikkumist, kui mitu voogu kirjutavad ühte faili, edastab AWK neile ühenduse failide nimed, kuhu andmeid kirjutada, näiteks, chr_10_bin_52_batch_2_aa.csv. Tulemusena luuakse kettale palju väikseid faile (selleks kasutasin terabaidiseid EBS-mahte).

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

R-i tõlkimine?

Mida ma õppisin: Saate sellele juurdepääsu stdin ja stdout R-i skriptist, seega saab seda kasutada ka voogus.

Bash-skriptis võisite sellist rida märgata: ...cat chunked/*_bin_{}_*.csv | .\/upload_as_rds.R.... See tõlgib kõik ühendatud failid grupist (bin) allolevasse R-skripti. {} on eriline meetod parallel, mis sisestab kõik selle määratud voogu saadetavad andmed otse käsusse. Valik {#} pakub unikaalset voogude täitmise ID-d, ja {%} on tööülesande pesade number (kordub, kuid mitte kunagi samaaegselt). Kõikide valikute loendi leiate 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, R-skripti tõlgitud andmed laaditakse raamistikku, mis seejärel salvestatakse .rds-failina kasutades aws.s3 salvestatakse otse S3-sse.

RDS on midagi nagu Parquet'i noorem versioon, ilma veergu salvestamise rafineerimiseta.

Bash-skripti lõpetamisel sain hulga .rds-faile, mis asusid S3-s, mis võimaldas mul kasutada efektiivset kokkusurumist ja sisseehitatud tüüpe.

Malest R-i kasutamine töötas väga kiiresti, kuigi see oli aeglane. Pole üllatav, et R-i andmete lugemise ja kirjutamise tükkide optimeerimine on hea. Pärast keskmise suurusega kromosoomi testimist lõppes ülesanne C5n.4xl instansil umbes kahe tunni jooksul.

S3 piirangud

Mida ma õppisin: Tänu nutikale S3 teede rakendusele saab see hallata palju faile.

Ma olin mures, kas S3 suudab hallata suurt hulka edastatud faile. Ma saaksin faili nimed teha mõistlikeks, kuid kuidas S3 neid otsib?

Kaustad S3-s on vaid ilu pärast, tegelikult ei huvita süsteemi sümbolid /. S3 KKK lehelt.

Tundub, et S3 esindab faili teed lihtsalt võtme kaudu omamoodi hajus- või dokumendibaasis. Baki (bucket) võib pidada tabeliks ja failid tabeli ridadeks.

Kuna kiirus ja efektiivsus on Amazonis kasumi teenimiseks olulised, ei ole üllatav, et see 'võti-faili-teele' süsteem on erakordselt optimeeritud. Olen püüdnud leida tasakaalu: et ei peaks tegema hulgaliselt get-päringuid, kuid samas päringud toimiksid kiiresti. Selgus, et kõige parem on teha umbes 20 000 bin-faili. Arvan, et kui jätkata optimeerimist, võiks kiirus veelgi tõusta (näiteks luua spetsiaalne bak, et andmed seal hoida, vähendades seeläbi otsingutabeli suurust). Kuid edasiste katsetustega ei olnud juba aega ega raha.

Kuidas on lood ristühilduvusega?

Mida ma õppisin: peamine põhjus, miks aega kaotatakse, on teie andmete salvestamise meetodi enneaegne optimeerimine.

Selles etapis on väga oluline küsida endalt: 'Miks kasutada omandiõiguse formaate?' Põhjus peitub laadimiskiirusest (gzippitud CSV-failid laadisid 7 korda kauem) ja meie tööprotsesside ühilduvusest. Võin oma otsuse uuesti läbi vaadata, kui R suudab hõlpsasti laadida Parquet (või Arrow) faile ilma Sparki koormuseta. Meie laboris kasutavad kõik R-i, ja kui mul on vaja andmeid muuda, siis mul on endiselt allikateks tekstifailid, seega võin lihtsalt toru uuesti käivitada.

Tööde jagamine

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

Olen vooliku üks kromosoomi tööprotsessi silunud, nüüd tuleb töödelda kõik ülejäänud andmed.
Soovisin tõsta mitu EC2 instantsi töötlemiseks, kuid ühtlasi kartsin saada äärmiselt tasakaalustamata koormust erinevates töötlemise ülesannetes (nagu ka Spark kannatas tasakaalustamata partitsioonide all). Lisaks ei olnud mul soov tõsta ühte instantsi igale kromosoomile, kuna AWS-i kontodele on vaikimisi piirang 10 instantsi.

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

Esialgu palusin S3-l arvestada, kui palju ruumi iga kromosoom salvestuses võtab.

library(aws.s3)
library(tidyverse)

chr_sizes % 
  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
         
 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

Siis kirjutasin funktsiooni, mis võtab kogusumma, segab kromosoomide järjekorra, jagab need gruppidesse num_jobs ja teatab, kui palju eristuvad kõik töötlemistööd suuruses.

num_jobs <- 7
# Kui suur oleks iga töö, kui jagada ideaalselt?
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)



# A 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 komplekti ülesandeid, mis olid väga sarnased suurusele. Seejärel jäi vaid oma eelmist Bash-skripti suure ringi alla panna for. Selle optimeerimise kirjutamiseks kulus umbes 10 minutit. Ja see on palju vähem, kui oleksin kulutanud käsitsi ülesannete loomisele, kui need ei oleks tasakaalus. Seega arvan, et sellega eelneva optimeerimisega ei läinud ma petta.

for DESIRED_CHR in "16" "9" "7" "21" "MT"
do
# Üksiku kromosoomi töötlemise kood
fi

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

sudo shutdown -h now

… ja see kõik õnnestus! AWS CLI abil tõstsin instantsid ja läbi valiku user_data edastasin neile Bash-skripte nende töötlemise ülesannete jaoks. Need täideti ja suleti automaatselt, nii et ma ei maksnud liigse arvutusvõimsuse eest.

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

Pakendame kokku!

Mida ma õppisin: API peab olema lihtne, et tagada lihtsus ja paindlikkus kasutamisel.

Lõpuks sain ma andmed soovitud kohas ja vormis. Jäänud oli protsessi maksimaalselt lihtsustada, et mu kolleegidel oleks kergem. Soovisin luua lihtsa API päringute tegemiseks. Kui tulevikus otsustan, et tahan üle minna .rds Parquet-failide puhul peaks see olema probleem minu jaoks, mitte kolleegide jaoks. Seetõttu otsustasin teha sisemise R-paketi.

Kogusin ja dokumenteerisin väga lihtsa paketi, mis sisaldab vaid mõnda funktsiooni andmete juurde pääsemiseks, mis on koondatud funktsiooni get_snp. Samuti tegin kolleegidele veebisaidi pkgdown, et nad saaksid hõlpsasti vaadata näiteid ja dokumentatsiooni.

Intelligentne vahemälu

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

Kuna üks peamistest töövoogudest rakendas paketi SNP jaoks sama analüüsimudelit, otsustasin kasutada rühmitamist (binning) oma huvides. SNP-de andmeid edastades on tagastatavale objektile lisatud kogu teave grupist (bin). See tähendab, et vanad päringud võivad (teoreetiliselt) uute päringute töötlemist kiirendada.

# 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 palju võrdlusi, et võrrelda erinevate meetodite kasutamise kiirus. Soovitan sellega mitte alahinnata, kuna tulemused võivad olla üllatavad. Näiteks dplyr::filter oli märgatavalt kiirem ridade jääktingitud filtreerimise kaudu haaramiseks ja ühe veeru saamine filtreeritud andmekaardist töötas palju kiiremini kui indekseerimise süntaks.

Pange tähele, et objekt prev_snp_results sisaldab võtit snps_in_bin. See on kõigi unikaalsete SNP-de massiiv grupis (bin), mis võimaldab kiiresti kontrollida, kas eelneva päringu andmed on juba olemas. Samuti hõlbustab see kõigi SNP-de rühma (bin) kaudu tsüklilist läbimist järgmise koodi abil:

# 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äitasid ainult nelja ebaolulise koodibloki kattuvust, mis olid tingitud POSIX ja ANSI C nõuetest.

Nüüd saame (ja oleme hakanud tõsiselt) jooksma mudeleid ja stsenaariume, millele meil varem ei olnud juurdepääsu. Parim asi on see, et mu kolleegidel laboratooriumis ei pea muretsema keerukuste üle. Neil on lihtsalt töötav funktsioon.

Ja kuigi pakett vabastab nad detailidest, üritasin teha andmeformaat piisavalt lihtsaks, et nad saaksid sellega toime tulla, kui ma homme äkki kadun …

Kiirus on märgatavalt kasvanud. Tavaliselt skaneerime funktsionaalselt olulisi geneetilisi fragmente. Varem ei saanud me seda teha (see oli liiga kallis), kuid nüüd, tänu grupistruktuurile (bin) ja vahemälule, kulub ühe SNP päringule keskmiselt vähem kui 0,1 sekundit ning andmete kasutamine on nii madal, et S3 kulud on tühised.

Hiljuti sain vastutada üle 25 TB toore genotüübiandmete haldamise eest oma laboris. Kui alustasin, nõudis spark SNP päringuks 8 minutit ja maksis $20. Pärast AWK + töötlemise kasutamist võtab see nüüd vähem kui 1/10 sekundi ja maksab $0,00001. Minu isiklik #rstats võit. #BigData pic.twitter.com/ANOXVGrmkk 30. mai 2019

— Nick Strayer (@NicholasStrayer) See artikkel ei ole juhend. Lahendus oli individuaalne ja tõenäoliselt mitte optimaalne. Pigem on see lugu reisist. Soovin, et teised mõistaksid, et sellised lahendused ei tule alati valmis mõtlema, vaid on katsete ja eksimiste tulemus. Lisaks, kui otsite andmeanalüüsi spetsialisti, pidage meeles, et nende tööriistade tõhusaks kasutamiseks on vajalik kogemus, mis omakorda nõuab raha. Olen rõõmus, et mul oli raha maksmiseks, kuid paljudel teistel, kes saaksid sama tööd paremini teha, ei ole kunagi sellist võimalust raha puudumise tõttu isegi katsetamiseks.

Kokkuvõte

Suured andmete töötlustooted on universaalsed. Kui teil on aega, võite peaaegu kindlasti kirjutada kiirema lahenduse, rakendades 'nutikat' andmete puhastamist, ladustamist ja ekstraktsioonimeetodeid. Lõppude lõpuks kõik sõltub kulude ja kasumi analüüsist.

Mida ma õppisin:

odavat viisi 25 TB korraga töötlemiseks ei ole;

• olge ettevaatlik oma Parquet-failide suuruse ja korraldusega;
• Sparkis peavad partiid olema tasakaalus;
• kunagi не üritage teha 2,5 miljonit partiid;
• sortimine on endiselt keeruline, nagu ka Spark'i seadistamine;
• mõnikord vajavad erilised andmed erilisi lahendusi;
• Sparkis toimib liitmine kiiresti, kuid partitseerimine on endiselt kulukas;
• ärge magage, kui õpetatakse aluseid, keegi on kindlasti teie probleemi juba 1980ndatel lahendanud;
• — see on maagiline asi, kõik peaks seda kasutama;
• gnu parallel Spark armastab kokkusurumata andmeid ja ei armasta partitsioonide ühendamist;
• Sparkis on liiga palju kulusid lihtsate ülesannete lahendamisel;
• assosiatiivsed massiivid AWK-s on väga efektiivsed;
• neile saab ligi pääseda
• R-skriptsist, seega saab seda kasutada ka torustikus; stdin ja stdout tänu nutikale S3 teede rakendusele suudab töödelda palju faile;
• aegade raiskamise peamine põhjus on teie ladustamisviisi enneaegne optimeerimine;
• ärge proovige optimeerida ülesandeid käsitsi, laske sellel teha arvuti;
• API peab olema lihtne, et tagada kasutamise lihtsus ja paindlikkus;
• kui teie andmed on hästi ette valmistatud, on vahemällu salvestamine lihtne!
• kui teie andmed on hästi ette valmistatud, on vahemälu salvestamine lihtne!

Allikas: habr.com