Hur man läser den här artikeln: Jag ber om ursäkt för att texten är så lång och kaotisk. För att spara tid börjar jag varje kapitel med en introduktion "Vad jag lärde mig", som sammanfattar kärnan i kapitlet i en eller två meningar.

"Visa mig bara lösningen!" Om du bara vill se var jag kom ifrån, hoppa till kapitlet "Bli mer uppfinningsrik", men jag tycker att det är mer intressant och användbart att läsa om misslyckanden.

Jag fick nyligen i uppdrag att sätta upp en process för att bearbeta en stor volym av råa DNA-sekvenser (tekniskt sett ett SNP-chip). Behovet var att snabbt få data om en given genetisk plats (kallad SNP) för efterföljande modellering och andra uppgifter. Med R och AWK kunde jag rengöra och organisera data på ett naturligt sätt, vilket avsevärt påskyndade frågebehandlingen. Detta var inte lätt för mig och krävde många iterationer. Den här artikeln hjälper dig att undvika några av mina misstag och visar dig vad jag slutade med.

Först några inledande förklaringar.

Data

Vårt center för behandling av genetisk information vid universitetet försåg oss med data i form av en 25 TB TSV. Jag fick dem uppdelade i 5 Gzip-komprimerade paket, som vart och ett innehöll cirka 240 fyra gigabyte filer. Varje rad innehöll data för en SNP från en individ. Totalt överfördes data om ~2,5 miljoner SNP:er och ~60 tusen människor. Förutom SNP-information innehöll filerna många kolumner med siffror som speglar olika egenskaper, såsom läsintensitet, frekvens av olika alleler, etc. Totalt fanns det cirka 30 kolumner med unika värden.

Mål

Som med alla datahanteringsprojekt var det viktigaste att bestämma hur datan skulle användas. I detta fall vi kommer mestadels att välja modeller och arbetsflöden för SNP baserat på SNP. Det vill säga att vi bara behöver data på en SNP åt gången. Jag var tvungen att lära mig hur man hämtar alla poster associerade med en av de 2,5 miljoner SNP:erna så enkelt, snabbt och billigt som möjligt.

Hur man inte gör detta

För att citera en passande klyscha:

Jag misslyckades inte tusen gånger, jag upptäckte bara tusen sätt att undvika att analysera en massa data i ett frågevänligt format.

Första försöket

Vad har jag lärt mig: Det finns inget billigt sätt att analysera 25 TB åt gången.

Efter att ha gått kursen "Avancerade metoder för Big Data Processing" vid Vanderbilt University var jag säker på att tricket låg i bagaget. Det kommer förmodligen att ta en timme eller två att ställa in Hive-servern för att köra igenom all data och rapportera resultatet. Eftersom vår data lagras i AWS S3 använde jag tjänsten Athena, som låter dig tillämpa Hive SQL-frågor på S3-data. Du behöver inte konfigurera/höja ett Hive-kluster, och du betalar också endast för den data du letar efter.

Efter att jag visade Athena mina data och dess format, körde jag några tester med frågor som detta:

select * from intensityData limit 10;

Och fick snabbt välstrukturerade resultat. Redo.

Tills vi försökte använda uppgifterna i vårt arbete...

Jag blev ombedd att ta fram all SNP-information för att testa modellen på. 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 tar betalt efter mängden data som hittas, 5 USD per terabyte. Så denna enda begäran kostade $20 och åtta minuters väntan. För att köra modellen på all data var vi tvungna att vänta i 38 år och betala 50 miljoner dollar. Det här var uppenbarligen inte lämpligt för oss.

Det var nödvändigt att använda parkett...

Vad har jag lärt mig: Var försiktig med storleken på dina Parquet-filer och deras organisation.

Jag försökte först åtgärda situationen genom att konvertera alla TSV till Parkettfiler. De är praktiska för att arbeta med stora datamängder eftersom informationen i dem lagras i kolumnform: varje kolumn ligger i sitt eget minne/disksegment, till skillnad från textfiler, där rader innehåller element i varje kolumn. Och om du behöver hitta något, läs bara den obligatoriska kolumnen. Dessutom lagrar varje fil ett intervall av värden i en kolumn, så om värdet du letar efter inte är i kolumnens intervall, kommer Spark inte att slösa tid på att skanna hela filen.

Jag körde en enkel uppgift AWS-lim att konvertera våra TSV till Parkett och släppte de nya filerna i Athena. Det tog ca 5 timmar. Men när jag körde förfrågan tog det ungefär lika lång tid och lite mindre pengar att slutföra. Faktum är att Spark, i ett försök att optimera uppgiften, helt enkelt packade upp en TSV-bit och lade den i sin egen Parkett-bit. Och eftersom varje bit var tillräckligt stor för att hålla alla register för många människor, innehöll varje fil alla SNP:er, så Spark var tvungen att öppna alla filer för att extrahera den information den behövde.

Intressant nog är Parquets standard (och rekommenderade) komprimeringstyp, snappy, inte delbar. Därför var varje executor fast i uppgiften att packa upp och ladda ner hela 3,5 GB dataset.

Låt oss förstå problemet

Vad har jag lärt mig: Sortering är svårt, speciellt om data är distribuerad.

Det verkade för mig att nu förstod jag kärnan i problemet. Jag behövde bara sortera data efter SNP-kolumn, inte efter personer. Sedan kommer flera SNP:er att lagras i en separat dataklump, och då kommer Parquets "smarta" funktion "öppna endast om värdet är inom intervallet" att visa sig i all ära. Tyvärr visade det sig vara en svår uppgift att sortera igenom miljarder rader utspridda över ett kluster.

Jag går på algoritmkurs på college: "Usch, ingen bryr sig om beräkningskomplexiteten för alla dessa sorteringsalgoritmer"

Jag försöker sortera på en kolumn i en 20TB #gnista tabell: "Varför tar det här så lång tid?" #DataScience kämpar.

— Nick Strayer (@NicholasStrayer) Mars 11, 2019

AWS vill definitivt inte ge en återbetalning på grund av skälet "Jag är en distraherad student". Efter att jag körde sortering på Amazon Glue, körde den i 2 dagar och kraschade.

Hur är det med partitionering?

Vad har jag lärt mig: Skiljeväggar i Spark måste vara balanserade.

Sedan kom jag på idén att partitionera data i kromosomer. Det finns 23 av dem (och flera till om man tar hänsyn till mitokondrie-DNA och omappade regioner).
Detta gör att du kan dela upp data i mindre bitar. Om du bara lägger till en rad till Spark-exportfunktionen i Glue-skriptet partition_by = "chr", då bör data delas upp i hinkar.

Genomet består av många fragment som kallas kromosomer.

Tyvärr fungerade det inte. Kromosomer har olika storlekar, vilket betyder olika mängder information. Detta innebär att uppgifterna som Spark skickade till arbetarna inte var balanserade och slutfördes långsamt eftersom vissa noder slutade tidigt och var inaktiva. Uppgifterna var dock klara. Men när man bad om en SNP orsakade obalansen återigen problem. Kostnaden för att bearbeta SNP på större kromosomer (det vill säga där vi vill hämta data) har bara minskat med ungefär en faktor 10. Mycket, men inte tillräckligt.

Tänk om vi delar upp det i ännu mindre delar?

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

Jag bestämde mig för att gå all out och partitionera varje SNP. Detta säkerställde att partitionerna var lika stora. DET VAR EN DÅLIG IDÉ. Jag använde Glue och lade till en oskyldig linje partition_by = 'snp'. Uppgiften startade och började utföras. En dag senare kollade jag och såg att det fortfarande inte fanns något skrivet till S3, så jag dödade uppgiften. Det ser ut som att Glue skrev mellanfiler till en dold plats i S3, många filer, kanske ett par miljoner. Som ett resultat kostade mitt misstag mer än tusen dollar och behagade inte min mentor.

Partitionering + sortering

Vad har jag lärt mig: Sortering är fortfarande svårt, liksom att trimma Spark.

Mitt sista försök att partitionera innebar att jag partitionerade kromosomerna och sedan sorterade varje partition. I teorin skulle detta påskynda varje fråga eftersom den önskade SNP-datan måste ligga inom några parkettbitar inom ett givet intervall. Tyvärr visade det sig vara en svår uppgift att sortera även partitionerade data. Som ett resultat bytte jag till EMR för ett anpassat 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')
  )

...uppgiften var dock fortfarande inte klar. Jag konfigurerade det på olika sätt: ökade minnesallokeringen för varje frågeexekutor, använde noder med en stor mängd minne, använde broadcastvariabler (sändningsvariabler), men varje gång visade det sig att dessa var halvmått, och gradvis började exekverarna att misslyckas tills allt stannade.

Uppdatering: så börjar det. pic.twitter.com/agY4GU2ru5

— Nick Strayer (@NicholasStrayer) Maj 15, 2019

Jag blir mer kreativ

Vad har jag lärt mig: Ibland kräver specialdata speciallösningar.

Varje SNP har ett positionsvärde. Detta är ett tal som motsvarar antalet baser längs dess kromosom. Detta är ett trevligt och naturligt sätt att organisera vår data. Först ville jag dela upp efter regioner av varje kromosom. Till exempel positioner 1 - 2000, 2001 - 4000, etc. Men problemet är att SNP inte är jämnt fördelade över kromosomerna, så gruppstorlekarna kommer därför att variera mycket.

Som ett resultat kom jag till en uppdelning av positioner i kategorier (rang). Med hjälp av redan nedladdade data körde jag en begäran om att få en lista över unika SNP, deras positioner och kromosomer. Sedan sorterade jag data inom varje kromosom och samlade in SNPs i grupper (bin) av en given storlek. Låt oss säga 1000 SNP vardera. Detta gav mig förhållandet SNP-till-grupp-per-kromosom.

Till slut gjorde jag grupper (bin) med 75 SNP:er, anledningen kommer att förklaras nedan.

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

Försök först med Spark

Vad har jag lärt mig: Gnistaggregering är snabb, men partitionering är fortfarande dyrt.

Jag ville läsa den här lilla (2,5 miljoner rader) dataramen i Spark, kombinera 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å Spark vet att den ska skicka dataramen till alla noder. Detta är användbart om informationen är liten och krävs för alla uppgifter. Annars försöker Spark vara smart och distribuerar data efter behov, vilket kan orsaka nedgångar.

Och återigen, min idé fungerade inte: uppgifterna fungerade under en tid, fullbordade facket, och sedan började de misslyckas, precis som exekutörerna som startade genom partitionering.

Lägger till AWK

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

Fram till denna punkt var orsaken till alla mina misslyckanden med Spark virrvarret av data i klustret. Kanske kan situationen förbättras med förbehandling. Jag bestämde mig för att försöka dela upp den råa textdatan i kolumner med kromosomer i hopp om att förse Spark med "förpartitionerade" data.

Jag sökte på StackOverflow efter hur man delar upp efter kolumnvärden och hittade så bra svar. Med AWK kan du dela upp en textfil efter kolumnvärden genom att skriva den i ett skript istället för att skicka resultaten till stdout.

Jag skrev ett Bash-manus för att testa det. Laddade ner en av de paketerade TSV:erna och packade sedan upp den med hjälp av gzip och skickas till awk.

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

Det fungerade!

Fyller kärnorna

Vad har jag lärt mig: gnu parallel – det är en magisk sak, alla borde använda den.

Separationen var ganska långsam och när jag började htopför att kontrollera användningen av en kraftfull (och dyr) EC2-instans visade det sig att jag bara använde en kärna och cirka 200 MB minne. För att lösa problemet och inte förlora en massa pengar var vi tvungna att ta reda på hur vi skulle parallellisera arbetet. Lyckligtvis i en helt fantastisk bok Datavetenskap vid kommandoraden Jag hittade ett kapitel av Jeron Janssens om parallellisering. Av det lärde jag mig om gnu parallel, en mycket flexibel metod för att implementera multithreading i Unix.

När jag startade partitioneringen med den nya processen var allt bra, men det fanns fortfarande en flaskhals - nedladdning av S3-objekt till disk var inte särskilt snabb och inte helt parallelliserad. För att fixa detta gjorde jag så här:

1. Jag fick reda på att det är möjligt att implementera S3-nedladdningssteget direkt i pipeline, vilket helt eliminerar mellanlagring på disk. Det betyder att jag kan undvika att skriva rådata till disk och använda ännu mindre, och därför billigare, lagring på AWS.
2. Team aws configure set default.s3.max_concurrent_requests 50 ökade kraftigt antalet trådar som AWS CLI använder (som standard finns det 10).
3. Jag bytte till en EC2-instans optimerad för nätverkshastighet, med bokstaven n i namnet. Jag har funnit att förlusten av processorkraft vid användning av n-instanser mer än kompenseras av ökningen i laddningshastighet. För de flesta uppgifter använde jag c5n.4xl.
4. Ändrats gzip på pigz, detta är ett gzip-verktyg som kan göra coola saker för att parallellisera den initialt icke-parallelliserade uppgiften att dekomprimera filer (detta hjälpte minst).

# 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 kombineras med varandra för att få allt att fungera väldigt snabbt. Genom att öka nedladdningshastigheterna och eliminera diskskrivningar kunde jag nu bearbeta ett paket på 5 terabyte på bara några timmar.

Det finns inget sötare än att se alla kärnor du betalar för på AWS används. Tack vare gnu-parallel kan jag packa upp och dela upp en 19 gig csv lika snabbt som jag kan ladda ner den. Jag kunde inte ens få gnista för att köra detta. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) Maj 17, 2019

Denna tweet borde ha nämnt "TSV". Ack.

Använder nyligen analyserad data

Vad har jag lärt mig: Spark gillar okomprimerad data och gillar inte att kombinera partitioner.

Nu fanns data i S3 i ett uppackat (läs: delat) och halvorganiserat format, och jag kunde återvända till Spark igen. En överraskning väntade mig: jag lyckades återigen inte uppnå det jag ville! Det var mycket svårt att berätta för Spark exakt hur data var uppdelad. Och även när jag gjorde detta visade det sig att det fanns för många partitioner (95 tusen), och när jag använde coalesce reducerade deras antal till rimliga gränser, detta förstörde min partitionering. Jag är säker på att detta går att fixa, men efter ett par dagars letande kunde jag inte hitta någon lösning. Jag avslutade så småningom alla uppgifter i Spark, även om det tog ett tag och mina delade parkettfiler var inte särskilt små (~200 KB). Däremot fanns uppgifterna där de behövdes.

För liten och ojämn, underbar!

Testar lokala Spark-frågor

Vad har jag lärt mig: Spark har för mycket omkostnader när man löser enkla problem.

Genom att ladda ner datan i ett smart format kunde jag testa hastigheten. Ställ in ett R-skript för att köra en lokal Spark-server och laddade sedan en Spark-dataram från den angivna Parquet-grupplagringen (bin). Jag försökte ladda all data men kunde inte få Sparklyr att känna igen partitioneringen.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Start a timer
start_time <- Sys.time()

# Load the desired bin into Spark
intensity_data <- sc %>% 
  Spark_read_Parquet(
    name = 'intensity_data', 
    path = get_snp_location(desired_snp),
    memory = FALSE )

# Subset bin to snp and then collect to local
test_subset <- intensity_data %>% 
  filter(SNP_Name == desired_snp) %>% 
  collect()

print(Sys.time() - start_time)

Avrättningen tog 29,415 sekunder. Mycket bättre, men inte för bra för masstestning av någonting. Dessutom kunde jag inte påskynda saker och ting med cachelagring eftersom när jag försökte cachelagra en dataram i minnet kraschade Spark alltid, även när jag allokerade mer än 50 GB minne till en datauppsättning som vägde mindre än 15.

Återgå till AWK

Vad har jag lärt mig: Associativa arrayer i AWK är mycket effektiva.

Jag insåg att jag kunde uppnå högre hastigheter. Jag kom ihåg det i en underbar AWK handledning av Bruce Barnett Jag läste om en cool funktion som heter "associativa arrayer" I huvudsak är dessa nyckel-värdepar, som av någon anledning kallades annorlunda i AWK, och därför tänkte jag på något sätt inte så mycket på dem. Roman Cheplyaka påminde om att termen "associativa arrayer" är mycket äldre än termen "nyckel-värdepar". Även om du slå upp nyckel-värdet i Google Ngram, du kommer inte att se den här termen där, men du hittar associativa arrayer! Dessutom är "nyckel-värde-paret" oftast associerat med databaser, så det är mycket mer meningsfullt att jämföra det med en hashmap. Jag insåg att jag kunde använda dessa associativa arrayer för att associera mina SNP:er med en bin-tabell och rådata utan att använda Spark.

För att göra detta använde jag blocket i AWK-skriptet BEGIN. Detta är en bit kod som exekveras innan den första raden med data skickas till skriptets huvuddel.

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

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

variabler batch_num и chunk_id matchade data som tillhandahålls av pipelinen, undviker ett race-tillstånd och varje exekveringstråd som körs parallel, skrev till sin egen unika fil.

Eftersom jag spred alla rådata i mappar på kromosomer som blev över från mitt tidigare experiment med AWK, kunde jag nu skriva ett annat Bash-skript för att bearbeta en kromosom i taget och skicka djupare partitionerad data till S3.

DESIRED_CHR='13'

# Download chromosome data from s3 and split into bins
aws s3 ls $DATA_LOC |
awk '{print $4}' |
grep 'chr'$DESIRED_CHR'.csv' |
parallel "echo 'reading {}'; aws s3 cp "$DATA_LOC"{} - | awk -v chr=""$DESIRED_CHR"" -v chunk="{}" -f split_on_chr_bin.awk"

# Combine all the parallel process chunks to single files and upload to rds using R
ls chunked/ |
cut -d '_' -f 4 |
sort -u |
parallel "echo 'zipping bin {}'; cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R '$S3_DEST'/chr_'$DESIRED_CHR'_bin_{}.rds"
rm chunked/*

Manuset har två avsnitt parallel.

I det första avsnittet läses data från alla filer som innehåller information om den önskade kromosomen, sedan distribueras denna data över trådar, som distribuerar filerna i lämpliga grupper (bin). För att undvika tävlingsförhållanden när flera trådar skriver till samma fil skickar AWK filnamnen för att skriva data till olika platser, t.ex. chr_10_bin_52_batch_2_aa.csv. Som ett resultat skapas många små filer på disken (för detta använde jag terabyte EBS-volymer).

Transportör från andra sektionen parallel går igenom grupperna (bin) och kombinerar deras individuella filer till gemensamma CSV c catoch skickar dem sedan för export.

Sändning i R?

Vad har jag lärt mig: Du kan kontakta stdin и stdout från ett R-skript, och använd det därför i pipelinen.

Du kanske har lagt märke till den här raden i ditt Bash-skript: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Den översätter alla sammanlänkade gruppfiler (bin) till R-skriptet nedan. {} är en speciell teknik parallel, som infogar all data som den skickar till den angivna strömmen direkt i själva kommandot. Alternativ {#} ger ett unikt tråd-ID och {%} representerar jobbplatsnumret (upprepas, men aldrig samtidigt). En lista över alla alternativ finns i dokumentation.

#!/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 en variabel file("stdin") överförs till readr::read_csv, data som översatts till R-skriptet laddas in i en ram, som sedan är i form .rds-fil med hjälp av aws.s3 skrivet direkt till S3.

RDS är ungefär en juniorversion av Parkett, utan krusidullerna med högtalarförvaring.

Efter att ha avslutat Bash-manuset fick jag ett paket .rds-filer som finns i S3, vilket gjorde att jag kunde använda effektiv komprimering och inbyggda typer.

Trots användningen av broms R fungerade allt väldigt snabbt. Inte överraskande är de delar av R som läser och skriver data mycket optimerade. Efter testning på en medelstor kromosom slutfördes jobbet på en C5n.4xl-instans på cirka två timmar.

S3 Begränsningar

Vad har jag lärt mig: Tack vare smart vägimplementering kan S3 hantera många filer.

Jag var orolig om S3 skulle kunna hantera de många filer som överfördes till den. Jag skulle kunna göra filnamnen vettiga, men hur skulle S3 leta efter dem?

Mappar i S3 är bara för att visa, i själva verket är systemet inte intresserad av symbolen /. Från S3 FAQ-sidan.

Det verkar som att S3 representerar sökvägen till en viss fil som en enkel nyckel i en sorts hashtabell eller dokumentbaserad databas. En hink kan ses som en tabell, och filer kan betraktas som poster i den tabellen.

Eftersom snabbhet och effektivitet är viktiga för att göra en vinst hos Amazon, är det ingen överraskning att detta nyckel-som-fil-sökvägssystem är galet optimerat. Jag försökte hitta en balans: så att jag inte behövde göra många get-förfrågningar, men att förfrågningarna utfördes snabbt. Det visade sig att det är bäst att göra cirka 20 tusen bin-filer. Jag tror att om vi fortsätter att optimera kan vi uppnå en ökning av hastigheten (till exempel genom att göra en speciell hink enbart för data och därmed minska storleken på uppslagstabellen). Men det fanns varken tid eller pengar för ytterligare experiment.

Hur är det med korskompatibilitet?

Vad jag lärde mig: Den främsta orsaken till bortkastad tid är att optimera din lagringsmetod i förtid.

Vid det här laget är det mycket viktigt att fråga dig själv: "Varför använda ett proprietärt filformat?" Anledningen ligger i laddningshastigheten (gzippade CSV-filer tog 7 gånger längre tid att ladda) och kompatibilitet med våra arbetsflöden. Jag kan ompröva om R enkelt kan ladda Parquet (eller Arrow) filer utan Spark-belastningen. Alla i vårt labb använder R, och om jag behöver konvertera data till ett annat format har jag fortfarande den ursprungliga textdatan, så jag kan bara köra pipelinen igen.

Uppdelning av arbete

Vad har jag lärt mig: Försök inte att optimera jobb manuellt, låt datorn göra det.

Jag har felsökt arbetsflödet på en kromosom, nu måste jag bearbeta alla andra data.
Jag ville ta upp flera EC2-instanser för konvertering, men samtidigt var jag rädd för att få en väldigt obalanserad belastning över olika bearbetningsjobb (precis som Spark led av obalanserade partitioner). Dessutom var jag inte intresserad av att höja en instans per kromosom, eftersom det för AWS-konton finns en standardgräns på 10 instanser.

Sedan bestämde jag mig för att skriva ett manus i R för att optimera bearbetningsjobb.

Först bad jag S3 att beräkna hur mycket lagringsutrymme varje kromosom upptog.

library(aws.s3)
library(tidyverse)

chr_sizes <- get_bucket_df(
  bucket = '...', prefix = '...', max = Inf
) %>% 
  mutate(Size = as.numeric(Size)) %>% 
  filter(Size != 0) %>% 
  mutate(
    # Extract chromosome from the file name 
    chr = str_extract(Key, 'chr.{1,4}.csv') %>%
             str_remove_all('chr|.csv')
  ) %>% 
  group_by(chr) %>% 
  summarise(total_size = sum(Size)/1e+9) # Divide to get value in GB



# A tibble: 27 x 2
   chr   total_size
   <chr>      <dbl>
 1 0           163.
 2 1           967.
 3 10          541.
 4 11          611.
 5 12          542.
 6 13          364.
 7 14          375.
 8 15          372.
 9 16          434.
10 17          443.
# … with 17 more rows

Sedan skrev jag en funktion som tar den totala storleken, blandar ordningen på kromosomerna, delar upp dem i grupper num_jobs och berättar hur olika storlekarna på alla bearbetningsjobb är.

num_jobs <- 7
# How big would each job be if perfectly split?
job_size <- sum(chr_sizes$total_size)/7

shuffle_job <- function(i){
  chr_sizes %>%
    sample_frac() %>% 
    mutate(
      cum_size = cumsum(total_size),
      job_num = ceiling(cum_size/job_size)
    ) %>% 
    group_by(job_num) %>% 
    summarise(
      job_chrs = paste(chr, collapse = ','),
      total_job_size = sum(total_size)
    ) %>% 
    mutate(sd = sd(total_job_size)) %>% 
    nest(-sd)
}

shuffle_job(1)



# A tibble: 1 x 2
     sd data            
  <dbl> <list>          
1  153. <tibble [7 × 3]>

Sedan gick jag igenom tusen shuffles med purrr och valde det bästa.

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

Så jag fick en uppsättning uppgifter som var väldigt lika i storlek. Sedan var det bara att linda in mitt tidigare Bash-manus i en stor loop for. Denna optimering tog cirka 10 minuter att skriva. Och detta är mycket mindre än jag skulle spendera på att manuellt skapa uppgifter om de var obalanserade. Därför tror jag att jag hade rätt med denna preliminära optimering.

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

I slutet lägger jag till kommandot shutdown:

sudo shutdown -h now

... och allt löste sig! Med hjälp av AWS CLI tog jag upp instanser med alternativet user_data gav dem Bash-skript av deras uppgifter för bearbetning. De körde och stängdes av automatiskt, så jag betalade inte för extra processorkraft.

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

Låt oss packa!

Vad har jag lärt mig: API:et bör vara enkelt för enkelhetens och flexibilitetens skull.

Äntligen fick jag uppgifterna på rätt plats och i rätt form. Allt som återstod var att förenkla processen att använda data så mycket som möjligt för att göra det enklare för mina kollegor. Jag ville göra ett enkelt API för att skapa förfrågningar. Om jag i framtiden bestämmer mig för att byta från .rds till Parkettfiler, så borde detta vara ett problem för mig, inte för mina kollegor. För detta bestämde jag mig för att göra ett internt R-paket.

Bygg och dokumentera ett mycket enkelt paket som innehåller bara några få dataåtkomstfunktioner organiserade runt en funktion get_snp. Jag gjorde även en hemsida för mina kollegor packdown, så att de enkelt kan se exempel och dokumentation.

Smart cachning

Vad har jag lärt mig: Om din data är väl förberedd blir caching enkelt!

Eftersom ett av huvudarbetsflödena tillämpade samma analysmodell på SNP-paketet, bestämde jag mig för att använda binning till min fördel. Vid överföring av data via SNP bifogas all information från gruppen (bin) till det returnerade objektet. Det vill säga, gamla frågor kan (i teorin) påskynda behandlingen av nya frågor.

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

När jag byggde paketet körde jag många benchmarks för att jämföra hastighet när jag använde olika metoder. Jag rekommenderar att inte försumma detta, för ibland är resultaten oväntade. Till exempel, dplyr::filter var mycket snabbare än att fånga rader med indexeringsbaserad filtrering, och att hämta en enda kolumn från en filtrerad dataram var mycket snabbare än att använda indexeringssyntax.

Observera att objektet prev_snp_results innehåller nyckeln snps_in_bin. Detta är en uppsättning av alla unika SNP:er i en grupp (bin), så att du snabbt kan kontrollera om du redan har data från en tidigare fråga. Det gör det också enkelt att gå igenom alla SNP:er i en grupp (bin) med denna 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 börjat på allvar) köra modeller och scenarier som tidigare var otillgängliga för oss. Det bästa är att mina labbkollegor inte behöver tänka på några komplikationer. De har bara en funktion som fungerar.

Och även om paketet besparar dem detaljerna, försökte jag göra dataformatet tillräckligt enkelt för att de skulle kunna lista ut det om jag plötsligt försvann imorgon...

Hastigheten har ökat märkbart. Vi skannar vanligtvis funktionellt signifikanta genomfragment. Tidigare kunde vi inte göra detta (det visade sig vara för dyrt), men nu, tack vare grupp (bin) struktur och caching, tar en begäran om en SNP i genomsnitt mindre än 0,1 sekunder, och dataanvändningen är så låg att kostnaderna för S3 är jordnötter.

Nyligen fick jag en förändring av att gräla över 25+ TB av rå genotypningsdata för mitt labb. När jag började tog det 8 minuter att använda spark och kostade $20 för att fråga en SNP. Efter att ha använt AWK+ #rstats att bearbeta tar det nu mindre än en tiondels sekund och kostar $10. Min personliga #Big data vinna. pic.twitter.com/ANOXVGrmkk

— Nick Strayer (@NicholasStrayer) Maj 30, 2019

Slutsats

Den här artikeln är inte alls en guide. Lösningen visade sig vara individuell, och nästan säkert inte optimal. Det är snarare en reseskildring. Jag vill att andra ska förstå att sådana beslut inte verkar helt formade i huvudet, de är resultatet av försök och misstag. Dessutom, om du letar efter en datavetare, kom ihåg att användningen av dessa verktyg effektivt kräver erfarenhet och erfarenhet kostar pengar. Jag är glad att jag hade möjlighet att betala, men många andra som kan göra samma jobb bättre än jag kommer aldrig att få möjlighet på grund av brist på pengar att ens prova.

Big data-verktyg är mångsidiga. Om du har tid kan du nästan säkert skriva en snabbare lösning med smarta tekniker för datarensning, lagring och extraktion. I slutändan handlar det om en kostnads-nyttoanalys.

Vad jag lärde mig:

• det finns inget billigt sätt att analysera 25 TB åt gången;
• var försiktig med storleken på dina Parquet-filer och deras organisation;
• Skiljeväggar i Spark måste vara balanserade;
• I allmänhet, försök aldrig att göra 2,5 miljoner partitioner;
• Sortering är fortfarande svårt, liksom att sätta upp Spark;
• ibland kräver speciella data speciella lösningar;
• Gnistaggregering är snabb, men partitionering är fortfarande dyrt;
• sov inte när de lär dig grunderna, någon har förmodligen löst ditt problem redan på 1980-talet;
• gnu parallel - det här är en magisk sak, alla borde använda den;
• Spark gillar okomprimerad data och gillar inte att kombinera partitioner;
• Spark har för mycket omkostnader när man löser enkla problem;
• AWK:s associativa arrayer är mycket effektiva;
• du kan kontakta stdin и stdout från ett R-skript, och använd det därför i pipelinen;
• Tack vare smart vägimplementering kan S3 bearbeta många filer;
• Den främsta anledningen till att slösa tid är att i förtid optimera din lagringsmetod;
• försök inte att optimera uppgifter manuellt, låt datorn göra det;
• API:et bör vara enkelt för enkelhetens och flexibilitetens skull;
• Om din data är väl förberedd blir caching enkelt!

Källa: will.com