ProHoster > Blogg > Administration > Vi parsar 25TB med AWK och R

Vi parsar 25TB med AWK och R

Vi parsar 25TB med AWK och R
Hur man lÀser den hÀr artikeln: jag ber om ursÀkt för att texten blev sÄ lÄng och rörig. För att spara er tid inleder jag varje kapitel med en introduktion "Vad jag har lÀrt mig", dÀr jag i en eller tvÄ meningar sammanfattar kapitlets kÀrna.

"Visa bara lösningen!" Om du bara vill se vad jag kom fram till, gÄ till kapitlet "Bli mer uppfinningsrik", men jag anser att det Àr mer intressant och nyttigt att lÀsa om misslyckanden.

Nyligen fick jag i uppdrag att stÀlla in en process för att hantera en stor volym av ursprungliga DNA-sekvenser (tekniskt sett Àr detta en SNP-chip). Jag behövde snabbt fÄ data om en given genetisk position (som kallas SNP) för efterföljande modellering och andra uppgifter. Med hjÀlp av R och AWK lyckades jag rengöra och organisera data pÄ ett naturligt sÀtt, vilket avsevÀrt ökade hastigheten pÄ förfrÄgningarna. Det var inte lÀtt och krÀvde mÄnga iterationer. Den hÀr artikeln kommer att hjÀlpa dig att undvika nÄgra av mina misstag och visa vad jag faktiskt Ästadkom.

Först nÄgra inledande förklaringar.

Data

VÄrt universitets centrum för bearbetning av genetisk information gav oss data i TSV-format pÄ 25 TB. Jag fick dem uppdelade i 5 paket, komprimerade med Gzip, dÀr varje paket innehöll cirka 240 fyragigabytesfiler. Varje rad innehöll data för en SNP för en person. Totalt överfördes data för ~2,5 miljoner SNP och ~60 000 personer. Utöver SNP-information fanns det mÄnga kolumner med siffror som Äterspeglade olika egenskaper, sÄsom lÀshastighet, frekvens av olika alleler, och sÄ vidare. Totalt fanns det cirka 30 kolumner med unika vÀrden.

Syftet

Som i alla datastyrningsprojekt var det viktigaste att bestÀmma hur data skulle anvÀndas. I det hÀr fallet kommer vi huvudsakligen att anpassa modeller och arbetsflöden för SNP baserat pÄ SNP. SÄ vi kommer samtidigt bara att behöva data för en enda SNP. Jag mÄste lÀra mig att sÄ enkelt, snabbt och billigt som möjligt extrahera alla poster som rör en av 2,5 miljoner SNP.

Hur man inte gör detta

Jag citerar ett passande klyscha:

Jag har inte misslyckats tusen gÄnger, jag har bara upptÀckt tusen sÀtt att inte parsa en mÀngd data i ett begÀrligt format.

Första försöket

Vad jag har lÀrt mig: det finns inget billigt sÀtt att parsa 25 TB pÄ en gÄng.

NÀr jag lyssnade pÄ kursen 'Avancerade metoder för storskalig databehandling' vid Vanderbilt University, var jag sÀker pÄ att det skulle ordna sig. Kanske tar det en till tvÄ timmar att stÀlla in Hive-servern för att bearbeta all data och rapportera resultatet. Eftersom vÄra data lagras i AWS S3 anvÀnde jag tjÀnsten Athena, som gör att man kan köra Hive SQL-frÄgor pÄ S3-data. Man behöver inte stÀlla in/starta ett Hive-kluster och betalar bara för den data man söker.

NÀr jag visade Athena mina data och deras format körde jag nÄgra tester med liknande frÄgor:

select * from intensityData limit 10;

Och fick snabbt vÀlstrukturerade resultat. Klart.

Tills vi försökte anvĂ€nda data i arbetet


Jag blev ombedd att hÀmta all information om SNP för att testa modellen. Jag körde frÄgan:


select * from intensityData 
where snp = 'rs123456';


och började vĂ€nta. Efter Ă„tta minuter och mer Ă€n 4 TB begĂ€rd data fick jag resultatet. Athena debiterar för den mĂ€ngd data som hittas, $5 per terabyte. SĂ„ denna enda frĂ„ga kostade $20 och Ă„tta minuters vĂ€ntan. För att köra modellen pĂ„ all data skulle det ta 38 Ă„r och kosta $50 miljoner. Uppenbart var det inte optimalt för oss.

Vi behövde anvÀnda Parquet...

Vad jag har lÀrt mig: var försiktig med storleken pÄ dina Parquet-filer och deras organisation.

Först försökte jag fixa situationen genom att konvertera alla TSV till Parquet-filer.De Àr praktiska för att arbeta med stora datamÀngder eftersom informationen lagras i kolumnformat: varje kolumn ligger i en egen segment av minnet/disken, till skillnad frÄn textfiler dÀr raderna innehÄller element frÄn varje kolumn. Och om man behöver hitta nÄgot, rÀcker det att lÀsa den nödvÀndiga kolumnen. Dessutom innehÄller varje fil i kolumnen ett vÀrdeomrÄde, sÄ om det sökta vÀrdet inte finns inom kolumnens omrÄde, kommer Spark inte att slösa tid pÄ att skanna hela filen.

Jag körde en enkel uppgift AWS Glue För att konvertera vÄra TSV-filer till Parquet och ladda upp nya filer till Athena tog det ungefÀr 5 timmar. Men nÀr jag körde frÄgan tog det ungefÀr lika lÄng tid och kostade lite mindre. Problemet Àr att Spark, i syfte att optimera uppgiften, helt enkelt packade upp en TSV-chunk och placerade den i sin egen Parquet-chunk. Eftersom varje chunk var tillrÀckligt stor och innehöll kompletta poster för mÄnga personer, fanns alla SNP i varje fil, sÄ Spark behövde öppna alla filer för att extrahera den nödvÀndiga informationen.

Det Ă€r intressant att den standard (och rekommenderade) komprimeringstypen i Parquet — snappy — inte Ă€r delbar (splitable). DĂ€rför fastnade varje executor i uppgiften att packa upp och ladda det kompletta datasetet pĂ„ 3,5 GB.

Vi parsar 25TB med AWK och R

Vi utreder problemet

Vad jag har lÀrt mig: att sortera Àr svÄrt, sÀrskilt nÀr datan Àr fördelad.

Jag trodde att jag nu förstod kÀrnan i problemet. Jag behövde bara sortera datan efter kolumnen SNP, inte efter personer. DÄ skulle flera SNP lagras i en separat chunk av data, och dÄ skulle Parquets 'intelligenta' funktion att 'öppna endast om vÀrdet finns inom intervallet' verkligen trÀda fram. TyvÀrr visade det sig vara en svÄr uppgift att sortera miljarder rader spridda över klustret.

AWS vill definitivt inte Äterbetala pengarna med argumentet 'Jag Àr en distrÀ student'. Efter att jag körde sorteringen pÄ Amazon Glue arbetade den i 2 dagar och avslutades med ett fel.

Vad sÀgs om partitionering?

Vad jag har lÀrt mig: partitioner i Spark mÄste vara balanserade.

Sedan kom jag pÄ idén att partitionera datan i kromosomer. Det finns 23 stycken (plus nÄgra om man rÀknar med mitokondriell DNA och icke-kodade omrÄden).
Detta skulle göra det möjligt att dela upp datan i mindre portioner. Om jag lÀgger till en rad i Spark-funktionen för export i Glue-skriptet partition_by = "chr", sÄ bör datan delas upp i buckets.

Vi parsar 25TB med AWK och R
Genomet bestÄr av mÄnga fragment som kallas kromosomer.

TyvÀrr fungerade det inte. Kromosomer har olika storlekar, vilket innebÀr att de ocksÄ har olika mÀngder information. Det betyder att uppgifterna som Spark skickade till arbetarna inte var balanserade och utfördes lÄngsamt, eftersom vissa noder avslutade tidigare och blev inaktiva. Men uppgifterna genomfördes. Men nÀr jag begÀrde en SNP blev obalansen Äterigen ett problem. Kostnaden för att bearbeta SNP i större kromosomer (dÀr vi vill hÀmta data) minskade bara med ungefÀr 10 gÄnger. Mycket, men inte tillrÀckligt.

Vad hÀnder om vi delar upp i Ànnu mindre partitioner?

Vad jag har lÀrt mig: Försök aldrig göra 2,5 miljoner partitioner.

Jag bestĂ€mde mig för att överdriva och partitionerade varje SNP. Detta garanterade lika stora partitioner. DET VAR EN DÅLIG IDÉ. Jag anvĂ€nde Glue och la till en oskyldig rad partition_by = 'snp'. Uppgiften startade och började köras. En dag senare kontrollerade jag och sĂ„g att ingenting hade skrivits till S3, sĂ„ jag dödade uppgiften. Det verkar som om Glue skrev mellanliggande filer pĂ„ en dold plats i S3, och mĂ„nga filer, kanske flera miljoner. Som resultat kostade mitt misstag mer Ă€n tusen dollar och glĂ€djde inte min handledare.

Partitionering + sortering

Vad jag har lÀrt mig: Att sortera Àr fortfarande svÄrt, liksom att konfigurera Spark.

Den sista partitioneringsförsöket innebar att jag partitionerade kromosomerna och sedan sorterade varje partition. I teorin skulle detta snabba upp varje begĂ€ran, eftersom de önskade data om SNP skulle befinna sig inom nĂ„gra Parquet-schunkar inom det angivna intervallet. TyvĂ€rr visade det sig vara svĂ„rt att sortera Ă€ven partitionerade data. Som ett resultat gick jag över till EMR för att skapa en anpassad kluster och anvĂ€nde Ă„tta kraftfulla instanser (C5.4xl) och Sparklyr för att skapa ett mer flexibelt arbetsflöde


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


men uppgiften blev Ă€ndĂ„ inte slutförd. Jag justerade pĂ„ alla möjliga sĂ€tt: ökade minnesallokeringen för varje frĂ„ga-exekverare, anvĂ€nde noder med större minneskapacitet, applicerade sĂ€ndningsvariabler, men varje gĂ„ng var det halvmĂ„tt, och sĂ„ smĂ„ningom började exekverare att krascha tills allt stannade.

Jag blir mer pÄhittig

Vad jag har lÀrt mig: ibland krÀver sÀrskilda data sÀrskilda lösningar.

Varje SNP har ett positionsvĂ€rde. Det hĂ€r talet motsvarar antalet baser som ligger lĂ€ngs dess kromosom. Det Ă€r ett bra och naturligt sĂ€tt att organisera vĂ„ra data. Först ville jag partitionera efter omrĂ„den pĂ„ varje kromosom. Till exempel positioner 1 – 2000, 2001 – 4000 osv. Men problemet Ă€r att SNP Ă€r ojĂ€mt fördelade över kromosomerna, sĂ„ gruppstorlekarna kommer att variera kraftigt.

Vi parsar 25TB med AWK och R

Som ett resultat kom jag fram till en uppdelning i kategori (rank) positioner. Med de redan inlÀsta uppgifterna körde jag en förfrÄgan för att fÄ en lista över unika SNP, deras positioner och kromosomer. Jag sorterade sedan data inom varje kromosom och samlade SNP i grupper (bin) av en viss storlek. LÄt oss sÀga, 1000 SNP. Detta gav mig en koppling mellan SNP och gruppen i kromosomen.

Till slut gjorde jag grupper (bin) av 75 SNP, vilket jag förklarar nedan.

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

Första försöket med Spark

Vad jag har lÀrt mig: sammanslagning i Spark fungerar snabbt, men partitioneringen Àr fortfarande dyr.

Jag ville lÀsa in denna lilla (2,5 miljoner rader) dataram i Spark, slÄ ihop den med rÄdata och sedan partitionera den efter den nyligen tillagda kolumnen. 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')
  )

Jag anvÀnde sdf_broadcast(), sÄ att Spark vet att han ska skicka dataramen till alla noder. Detta Àr anvÀndbart om data Àr av liten storlek och behövs för alla uppgifter. Annars försöker Spark vara smart och distribuerar data efter behov, vilket kan orsaka flaskhalsar.

Och Äterigen fungerade min plan inte: uppgifterna kördes ett tag, avslutade sammanslagningen, och sedan, precis som de partitionerade exekverarna, började de krascha.

Jag lÀgger till AWK

Vad jag har lÀrt mig: sov inte nÀr du lÀr dig grunderna. NÄgon har sÀkert redan löst ditt problem pÄ 1980-talet.

Fram till den hÀr punkten har orsaken till alla mina misslyckanden med Spark varit oordning i klustret. Kanske kan situationen förbÀttras med hjÀlp av förbearbetning. Jag beslutade att försöka dela upp rÄdata i kromosomkolumner, sÄ jag hoppades kunna ge Spark "förpartitionerade" data.

Jag sökte pÄ StackOverflow hur man delar upp efter kolumnvÀrden och hittade detta fantastiska svar. Med AWK kan du dela en textfil baserat pÄ kolumnvÀrden genom att göra en inskrivning i skriptet istÀllet för att skicka resultaten till stdout.

För att testa skrev jag ett Bash-skript. Jag laddade ner en av de packade TSV-filerna och packade upp den med hjÀlp av gzip och skickade till awk.

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

Det fungerade!

KĂ€rnorna fylls

Vad jag har lÀrt mig: gnu parallel Àr en magisk grej, alla borde anvÀnda det.

Dela processen gick ganska lÄngsamt, och nÀr jag körde htop, för att kontrollera anvÀndningen av den kraftfulla (och dyra) EC2-instansen, visade det sig att jag anvÀnde endast en kÀrna och ungefÀr 200 Mb minne. För att lösa problemet och inte förlora en massa pengar behövde jag komma pÄ hur man kan köra arbetet parallellt. Lyckligtvis i den helt fantastiska boken Data Science at the Command Line av Jeroen Janssens hittade jag ett kapitel som handlade om parallellisering. DÀr lÀrde jag mig om gnu parallel, en mycket flexibel metod för att implementera flertrÄdad bearbetning i Unix.

Vi parsar 25TB med AWK och R
NĂ€r jag körde delningen med hjĂ€lp av den nya processen var allt perfekt, men det fanns fortfarande en flaskhals – nedladdningen av S3-objekt till disken gick inte sĂ€rskilt snabbt och var inte helt parallelliserad. För att Ă„tgĂ€rda detta gjorde jag följande:

  1. Jag insÄg att jag kan implementera S3-nedladdningssteget direkt i pipelinen, vilket helt utesluter mellanlagring pÄ disken. Det betyder att jag kan undvika att skriva rÄdata till disken och anvÀnda en Ànnu mindre och dÀrför billigare lagring pÄ AWS.
  2. Kommandot aws configure set default.s3.max_concurrent_requests 50 ökade kraftigt antalet trÄdar som AWS CLI anvÀnder (som standard Àr det 10).
  3. Gick över till en EC2-instans optimerad för nÀtverkshastighet, med bokstaven n i namnet. Jag upptÀckte att berÀkningskraftsförlusten nÀr jag anvÀnde n-instanser kompenserades med rÄge av ökad nedladdningshastighet. För de flesta uppgifter anvÀnde jag c5n.4xl.
  4. Bytte ut gzip till pigz, ett gzip-verktyg som kan göra fantastiska saker för att parallellisera en ursprungligen icke-parallelliserad uppgift för att packa upp filer (detta hjÀlpte minst av allt). 

# 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

Dessa steg kombinerades med varandra sÄ att allt fungerade vÀldigt snabbt. Tack vare den ökade nedladdningshastigheten och avskaffandet av skrivning pÄ disken kunde jag nu bearbeta ett 5-terabyte paket pÄ bara nÄgra timmar.

Den hÀr tweeten skulle ha nÀmnt 'TSV'. TyvÀrr.

ÅteranvĂ€ndning av omparsat data

Vad jag har lÀrt mig: Spark Àlskar ocompressad data och ogillar att kombinera partitioner.

Nu lÄg datan i S3 i en icke-komprimerad (lÀs delad) och halvordnad format, och jag kunde ÄtervÀnda till Spark. Jag blev överraskad: jag kunde Äterigen inte fÄ det att fungera som jag ville! Det var mycket svÄrt att berÀtta för Spark hur datan var partitionerad. Och Àven nÀr jag gjorde det, visade det sig att det fanns för mÄnga partitioner (95 000), och nÀr jag med hjÀlp av coalesce minskade antalet till rimliga grÀnser, förstörde det min partitionering. Jag Àr sÀker pÄ att detta kan ÄtgÀrdas, men efter flera dagar av sökande kunde jag inte hitta en lösning. Till slut gjorde jag klart alla uppgifter i Spark, Àven om det tog lite tid, och mina delade Parquet-filer var inte sÄ smÄ (~200 kB). Men datan lÄg dÀr den skulle.

Vi parsar 25TB med AWK och R
För smÄ och olika, underbart!

Testa lokala Spark-förfrÄgningar

Vad jag har lÀrt mig: det Àr för mycket overhead i Spark för att lösa enkla uppgifter.

Efter att ha laddat upp data i ett genomtÀnkt format, kunde jag testa hastigheten. Jag stÀllde in ett skript i R för att köra en lokal Spark-server och laddade sedan in ett Spark-dataframe frÄn det angivna Parquet-grupplagret (bin). Jag försökte ladda alla data, men kunde inte fÄ Sparklyr att kÀnna igen partitioneringen.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Starta en timer
start_time <- Sys.time()

# Ladda den önskade bin i Spark
intensity_data % 
  Spark_read_Parquet(
    name = 'intensity_data', 
    path = get_snp_location(desired_snp),
    memory = FALSE )

# Dela bin till snp och sedan samla in till lokalt
test_subset % 
  filter(SNP_Name == desired_snp) %>% 
  collect()

print(Sys.time() - start_time)

Exekveringen tog 29,415 sekunder. Mycket bÀttre, men inte tillrÀckligt bra för massatestning av nÄgot. Dessutom kunde jag inte snabba upp det med caching, för nÀr jag försökte cacha dataframe i minnet, kraschade Spark alltid, Àven nÀr jag tilldelade mer Àn 50 GB minne för en dataset som vÀgde mindre Àn 15.

ÅtervĂ€nda till AWK

Vad jag har lÀrt mig: associativa arrayer i AWK Àr mycket effektiva.

Jag insĂ„g att jag kunde uppnĂ„ högre hastighet. Jag kom ihĂ„g att i den fantastiska handboken om AWK av Bruce Barnett Jag lĂ€ste om en cool funktion som kallas «associativa arrayer». I huvudsak Ă€r det nyckel-vĂ€rde-par som av nĂ„gon anledning kallas nĂ„got annat i AWK, dĂ€rför har jag inte riktigt tĂ€nkt pĂ„ dem. Roman Cheplyaka pĂ„minde mig om att termen «associativa arrayer» Ă€r mycket Ă€ldre Ă€n termen «nyckel-vĂ€rde-par». Även om du söker efter nyckel-vĂ€rde-par i Google Ngram, kommer du inte att se den termen dĂ€r, men du kommer att hitta associativa arrayer! Dessutom associeras «nyckel-vĂ€rde-par» oftast med databaser, sĂ„ det Ă€r mycket logiskt att jĂ€mföra med hashmap. Jag insĂ„g att jag kan anvĂ€nda dessa associativa arrayer för att koppla mina SNP med bin-tabellen och rĂ„data utan att anvĂ€nda Spark.

För detta anvÀnde jag blocket i AWK-skriptet BEGIN. Detta Àr en kodsnutt som körs innan den första dataraden överförs till huvuddelen av skriptet.

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

Team while(getline...) laddade alla rader frÄn gruppens CSV (bin), satte första kolumnen (namnet pÄ SNP) som nyckel för den associativa arrayen bin och det andra vÀrdet (gruppen) som vÀrde. Sedan i blocket { }, som körs för varje rad i huvudfilen, skickas varje rad till utdatafilen, som fÄr ett unikt namn beroende pÄ dess grupp (bin): ..._bin_"bin[$1]"_....

Variabler batch_num och chunk_id svarade mot data som tillhandahölls av pipelinen, vilket gjorde att man kunde undvika race conditions, och varje exekveringstrÄd som startades parallel, skrev till sin egen unika fil.

Eftersom jag delade upp all rÄdata i mappar baserat pÄ kromosomer frÄn mitt tidigare experiment med AWK, kunde jag nu skriva ett annat Bash-skript för att bearbeta en kromosom Ät gÄngen och skicka mer djupt partitionerade data till S3.

DESIRED_CHR='13'

# Ladda ner kromosomdata frÄn s3 och dela upp i bin
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"

# Kombinera alla parallella processer till enskilda filer och ladda upp till rds med 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/*

I skriptet finns tvÄ avsnitt parallel.

I det första avsnittet lÀses data frÄn alla filer som innehÄller information om den aktuella kromosomen, och dessa data fördelas över trÄdar som sprider filer i motsvarande grupper (bin). För att undvika tillstÄnd av tÀvling, dÀr flera trÄdar skriver till en fil, skickar AWK filnamn för att skriva data pÄ olika platser, till exempel, chr_10_bin_52_batch_2_aa.csv. Resultatet blir en mÀngd smÄ filer pÄ disken (för detta anvÀnde jag terabyte stora EBS-volymer).

Pipelinen frÄn det andra avsnittet parallel gÄr igenom grupperna (bin) och förenar deras separata filer till gemensamma CSV-filer med cat, och skickar dem sedan för export.

Överföring till R?

Vad jag har lÀrt mig: man kan hÀnvisa till stdin och stdout frÄn R-skript, vilket gör att man kan anvÀnda det i pipelinen.

I Bash-skriptet kan du ha lagt mÀrke till en sÄdan rad: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Den överför alla sammanhÀngande filer frÄn gruppen (bin) till det nedanstÄende R-skriptet. {} Àr en speciell metodik parallel, som direkt skickar all data som den sÀnder i den angivna strömmen till sjÀlva kommandot. Alternativet {#} tillhandahÄller ett unikt ID för körströmmar, och {%} Àr nummer pÄ uppdragets slot (upprepas men aldrig samtidigt). En lista över alla alternativ kan hittas i dokumentationen.

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

NÀr variabeln file("stdin") överlÀmnas till readr::read_csv, laddas data överfört till R-skriptet in i en dataram som sedan i form av .rds-fil skrivs direkt till S3 via aws.s3. RDS Àr nÄgot som en enklare version av Parquet, utan kolumnformatets finesser.

Efter att Bash-skriptet avslutades fick jag en uppsÀttning

-filer som lÄg i S3, vilket gav mig möjlighet att anvÀnda effektiv komprimering och inbyggda typer. .rdsTrots den tröga R, fungerade allt vÀldigt snabbt. Det Àr inte förvÄnande att de avsnitt pÄ R som ansvarar för att lÀsa och skriva data Àr vÀl optimerade. Efter testning pÄ en kromosom av medelstorlek, avslutades uppdraget pÄ en C5n.4xl-instans pÄ ungefÀr tvÄ timmar.

S3:s begrÀnsningar

: tack vare smart implementering av S3-vÀgarna kan den hantera mÄnga filer.

Vad jag har lÀrt migJag var orolig för om S3 skulle kunna hantera mÄnga filer som skickats till den. Jag kunde ge filnamnen en meningsfull struktur, men hur skulle S3 söka igenom dem?

Mappar i S3 Àr bara för utseendet, systemet bryr sig faktiskt inte om tecknet

Vi parsar 25TB med AWK och R
FrÄn S3:s FAQ-sida. /. FrÄn FAQ-sidan för S3.

Det verkar som att S3 representerar sökvÀgen till en specifik fil som en enkel nyckel i en slags hash-tabell eller dokumentbaserad databas. En bucket kan ses som en tabell, medan filer Àr poster i denna tabell.

Eftersom hastighet och effektivitet Àr viktiga för att generera vinst pÄ Amazon, Àr det inte förvÄnande att detta "nyckel-till-sökvÀg-till-fil"-system Àr fantastiskt optimerat. Jag försökte hitta en balans: sÄ att det inte behövdes göras mÄnga GET-förfrÄgningar, men ÀndÄ att förfrÄgningarna gick snabbt. Det visade sig att det var bÀst att göra cirka 20 000 bin-filer. Jag tror att om man fortsÀtter att optimera sÄ kan man öka hastigheten (till exempel göra en speciell bucket endast för data, vilket minskar storleken pÄ sök-tabellen). Men det fanns inte tid eller pengar för fler experiment.

Vad gÀller korskompatibilitet?

Vad jag har lÀrt mig: Huvudorsaken till tidsförlust Àr för tidig optimering av din lagringsmetod.

I det hÀr ögonblicket Àr det viktigt att stÀlla sig sjÀlv frÄgan: "Varför anvÀnda ett proprietÀrt filformat?" Anledningen ligger i hastigheten vid inlÀsning (komprimerade gzip CSV-filer laddades 7 gÄnger lÄngsammare) och kompatibiliteten med vÄra arbetsflöden. Jag kan ompröva mitt beslut om R enkelt kan ladda Parquet-filer (eller Arrow) utan belastning i form av Spark. I vÄrt labb anvÀnder alla R, och om jag behöver konvertera data till ett annat format har jag fortfarande de ursprungliga textdata, sÄ jag kan enkelt köra pipelinen igen.

Arbetets uppdelning

Vad jag har lÀrt mig: Försök inte optimera uppgifter manuellt, lÄt datorn göra det.

Jag har finjusterat arbetsflödet pÄ en kromosom, nu behöver jag bearbeta alla andra data.
Jag ville starta flera EC2-instanser för konverteringen, men jag var samtidigt orolig för att fÄ en extremt obalanserad belastning pÄ olika bearbetningsuppgifter (precis som Spark lidit av obalanserade partitioner). Dessutom ville jag inte starta en instans för varje kromosom, eftersom det finns en standardgrÀns pÄ 10 instanser för AWS-konton.

DÄ bestÀmde jag mig för att skriva ett skript i R för att optimera bearbetningsuppgifterna.

Först bad jag S3 att berÀkna hur mycket lagringsutrymme varje kromosom tar.

library(aws.s3)
library(tidyverse)

chr_sizes % 
  mutate(Size = as.numeric(Size)) %>% 
  filter(Size != 0) %>% 
  mutate(
    # Extrahera kromosom frÄn filnamnet 
    chr = str_extract(Key, 'chr.{1,4}.csv') %>%
             str_remove_all('chr|.csv')
  ) %>% 
  group_by(chr) %>% 
  summarise(total_size = sum(Size)/1e+9) # Dela för att fÄ vÀrdet i GB



# En 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.
# 
 med 17 fler rader

Sedan skrev jag en funktion som tar totalstorleken, blandar kromosomordningen och delar dem i grupper. num_jobs och rapporterar hur mycket storleken pÄ alla behandlingsjobb skiljer sig.

num_jobs <- 7
# Hur stor skulle varje jobb vara om det delades perfekt?
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)



# En tibble: 1 x 2
     sd data            
             
1  153.

Sedan körde jag tusen blandningar med hjÀlp av purrr och valde den bÀsta.

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

SÄ jag fick ett set av jobb som var mycket lika i storlek. DÄ Äterstod det bara att omsluta mitt tidigare Bash-skript i en stor loop. for. Det tog mig cirka 10 minuter att skriva denna optimering. Och det Àr betydligt mindre Àn vad jag skulle ha spenderat pÄ att manuellt skapa jobb i fall av obalans. SÄ jag tycker att denna preliminÀra optimering var en bra idé.

för DESIRED_CHR in "16" "9" "7" "21" "MT"
do
# Kod för att bearbeta en enskild kromosom
fi

I slutet lÀgger jag till en avstÀngningskommandon:

sudo shutdown -h now


 och allt fungerade! Jag anvĂ€nde AWS CLI för att starta instanser och skickade dem Bash-skripten för deras bearbetningsjobb via optionen user_data de körde och stĂ€ngdes av automatiskt, sĂ„ jag betalade inte för överflödig berĂ€kningskraft.

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

Nu paketerar vi!

Vad jag har lÀrt mig: API:et ska vara enkelt för enkelhetens och anvÀndbarhetens skull.

Äntligen fick jag data pĂ„ rĂ€tt plats och i rĂ€tt format. Det Ă„terstod bara att förenkla anvĂ€ndningen av data, sĂ„ mina kollegor skulle ha det lĂ€ttare. Jag ville skapa ett enkelt API för att göra förfrĂ„gningar. Om jag i framtiden bestĂ€mmer mig för att byta med .rds NĂ€r det gĂ€ller Parquet-filer bör detta vara ett problem för mig och inte för kollegorna. DĂ€rför beslutade jag mig för att skapa ett internt R-paket.

Jag har samlat och dokumenterat ett mycket enkelt paket som innehÄller bara nÄgra fÄ funktioner för att fÄ tillgÄng till data, centrerat kring funktionen get_snp. Jag har ocksÄ skapat en webbplats för kollegorna pkgdown, sÄ att de enkelt kan se exempel och dokumentation.

Vi parsar 25TB med AWK och R

Intelligent caching

Vad jag har lÀrt mig: om dina data Àr vÀl förberedda blir det enkelt att cacha!

Eftersom en av de viktigaste arbetsflödena applicerade samma analysmodell pÄ SNP-paketet, beslutade jag mig för att anvÀnda gruppindelning (binning) till min fördel. NÀr data överförs via SNP bifogas all information frÄn gruppen (bin) till det returnerade objektet. Det vill sÀga, gamla förfrÄgningar kan (i teorin) snabba upp behandlingen av nya förfrÄgningar.

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

Vid sammanstÀllningen av paketet körde jag mÄnga benchmark-tester för att jÀmföra hastigheten med olika metoder. Jag rekommenderar att inte förbise detta, eftersom resultaten ibland kan vara ovÀntade. Till exempel, dplyr::filter visade sig vara mycket snabbare Àn att fÄnga rader med hjÀlp av indexeringsbaserad filtrering, och att fÄ en kolumn frÄn den filtrerade dataramverk fungerade mycket snabbare Àn att anvÀnda indexeringssyntax.

Observera att objektet prev_snp_results innehÄller nyckeln snps_in_bin. Detta Àr en array av alla unika SNP i gruppen (bin), vilket möjliggör snabb kontroll av om data frÄn en tidigare förfrÄgan redan finns. Det förenklar ocksÄ den cykliska genomgÄngen av alla SNP i gruppen (bin) med följande kod:

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

Resultat

Nu kan vi (och har verkligen börjat) köra modeller och scenarier som tidigare var otillgÀngliga för oss. Det bÀsta av allt Àr att mina kollegor i labbet inte behöver tÀnka pÄ nÄgra svÄrigheter. De har bara en fungerande funktion.

Och Àven om paketet befriar dem frÄn detaljerna, försökte jag göra dataformatet tillrÀckligt enkelt sÄ att de kan förstÄ det om jag plötsligt skulle försvinna imorgon...

Hastigheten har ökat betydligt. Vanligtvis skannar vi funktionellt betydelsefulla fragment av genomet. Tidigare kunde vi inte göra detta (det var för dyrt), men nu, tack vare gruppstrukturen (bin) och caching, tar det i genomsnitt mindre Àn 0,1 sekund per förfrÄgan om en SNP, och anvÀndningen av data Àr sÄ lÄg att kostnaderna för S3 blir dÀrmed vÀldigt lÄga.

Slutsats

Den hÀr artikeln Àr inte alls en handledning. Lösningen blev individuell och Àr nÀstan sÀkert inte optimal. Snarare Àr det en berÀttelse om en resa. Jag vill att andra ska förstÄ att sÄdana lösningar inte dyker upp fullt formade i huvudet, utan Àr resultatet av försök och misstag. Dessutom, om du letar efter en specialist inom dataanalys, bör du tÀnka pÄ att för att effektivt anvÀnda dessa verktyg krÀvs erfarenhet, och erfarenhet kostar pengar. Jag Àr glad att jag hade medel att betala, men mÄnga andra som kan göra samma jobb bÀttre Àn jag, kommer aldrig ha den möjligheten pÄ grund av bristen pÄ pengar för ens ett försök.

Verktyg för stora data Àr mÄngsidiga. Om du har tid kan du nÀstan sÀkert skriva en snabbare lösning genom att anvÀnda "smart" datarengöring, lagring och extraktionsmetoder. I slutÀndan handlar allt om kostnads- och nyttoanalys.

Vad jag lÀrde mig:

  • det finns inget billigt sĂ€tt att parsa 25 TB pĂ„ en gĂ„ng;
  • var försiktig med storleken pĂ„ dina Parquet-filer och deras organisation;
  • partitionerna i Spark bör vara balanserade;
  • försök aldrig att göra 2,5 miljoner partitioner;
  • att sortera Ă€r fortfarande svĂ„rt, precis som att stĂ€lla in Spark;
  • ibland krĂ€ver speciella data speciella lösningar;
  • sammanfogning i Spark fungerar snabbt, men partitionering Ă€r fortfarande kostsamt;
  • sov inte nĂ€r grunderna undervisas, nĂ„gon har sĂ€kert redan löst ditt problem under 1980-talet;
  • gnu parallel det Ă€r en magisk sak, alla borde anvĂ€nda den;
  • Spark Ă€lskar okomprimerade data och gillar inte att kombinera partitioner;
  • det finns för mycket overhead i Spark nĂ€r man löser enkla uppgifter;
  • associativa arrayer i AWK Ă€r mycket effektiva;
  • kan nĂ„s frĂ„n R-skript, vilket betyder att det kan anvĂ€ndas i en pipeline; stdin och stdout tack vare smart implementation av S3-vĂ€gar kan mĂ„nga filer bearbetas;
  • den största anledningen till tidsförlust Ă€r för tidig optimering av din lagringsmetod;
  • försök inte att optimera uppgifter manuellt, lĂ„t datorn göra det;
  • API:t mĂ„ste vara enkelt för enkelhetens och anvĂ€ndarens flexibilitet;
  • om dina data Ă€r vĂ€l förberedda, blir det enkelt att cacha!
  • om dina data Ă€r vĂ€l förberedda blir det enkelt att cachera!

KĂ€lla: habr.com

Köp pĂ„litlig webbhotell med DDoS-skydd, VPS VDS-servrar đŸ”„ Köp pĂ„litlig webbhotell med DDoS-skydd, VPS VDS-servrar | ProHoster