Come leggere questo articolo: mi scuso che il testo risulti così lungo e caotico. Per risparmiare il vostro tempo, inizio ogni capitolo con un'introduzione "Cosa ho imparato", in cui riassumo in una o due frasi il contenuto del capitolo.

"Mostrami solo la soluzione!" Se volete semplicemente vedere dove sono arrivato, passate al capitolo "Divento più inventivo", ma penso che sia più interessante e utile leggere riguardo ai fallimenti.

Di recente mi è stato chiesto di configurare il processo di elaborazione di un ampio volume di sequenze di DNA (tecnicamente si tratta di un chip SNP). Dovevo ottenere rapidamente dati su una specifica posizione genetica (chiamata SNP) per la successiva modellazione e altri compiti. Con R e AWK sono riuscito a pulire e organizzare i dati in modo naturale, accelerando notevolmente l'elaborazione delle richieste. Non è stato facile e ha richiesto numerose iterazioni. Questo articolo vi aiuterà ad evitare alcuni dei miei errori e mostrerà ciò che alla fine sono riuscito a ottenere.

Per iniziare, alcune spiegazioni introduttive.

Dati

Il nostro centro universitario di elaborazione delle informazioni genetiche ci ha fornito dati in formato TSV di 25 TB. Li ho ricevuti suddivisi in 5 pacchetti compressi Gzip, ognuno contenente circa 240 file da quattro gigabyte. Ogni riga conteneva dati per un SNP di una persona. In totale sono stati trasmessi dati su circa 2,5 milioni di SNP e 60.000 persone. Oltre alle informazioni sugli SNP, i file contenevano numerose colonne con numeri che riflettevano diverse caratteristiche, come l'intensità di lettura, la frequenza di diversi alleli, ecc. C'erano circa 30 colonne con valori unici.

Obiettivo

Come in ogni progetto di gestione dei dati, la cosa più importante era determinare come sarebbero stati utilizzati i dati. In questo caso principalmente svilupperemo modelli e flussi di lavoro per SNP basati sugli SNP. Questo significa che avremo bisogno dei dati solo su un SNP alla volta. Dovevo imparare a estrarre tutti i record relativi a uno dei 2,5 milioni di SNP in modo semplice, veloce ed economico.

Come non farlo

Citerò un cliché appropriato:

Non ho fallito mille volte, ho solo aperto mille modi per non elaborare una grande quantità di dati in un formato utile per le query.

Primo tentativo

Cosa ho imparato: non esiste un modo economico per analizzare 25 TB di dati in una sola volta.

Dopo aver seguito il corso "Metodi avanzati di elaborazione dei big data" presso l'Università di Vanderbilt, ero sicuro che sarebbe stato semplice. Probabilmente ci sarebbero volute un paio d'ore per configurare il server Hive e scorrere tutti i dati per presentare un risultato. Poiché i nostri dati sono archiviati in AWS S3, ho utilizzato il servizio Athena, che consente di applicare query SQL di Hive ai dati S3. Non è necessario configurare/sollevare un cluster Hive e paghi solo per i dati che cerchi.

Dopo aver mostrato i miei dati e il loro formato ad Athena, ho eseguito alcuni test con query simili:

select * from intensityData limit 10;

E ho rapidamente ottenuto risultati ben strutturati. Fatto.

Finché non abbiamo provato a utilizzare i dati nel lavoro…

Mi hanno chiesto di estrarre tutte le informazioni sui SNP per testare il modello. Ho lanciato la query:

select * from intensityData 
where snp = 'rs123456';

…e ho iniziato ad aspettare. Dopo otto minuti e oltre 4 TB di dati richiesti, ho ottenuto un risultato. Athena addebita in base al volume dei dati trovati, a $5 per terabyte. Quindi, questa singola richiesta è costata $20 e otto minuti di attesa. Per eseguire il modello su tutti i dati, sarebbe servito aspettare 38 anni e pagare $50 milioni. Ovviamente, questa soluzione non era adatta a noi.

Era necessario usare Parquet…

Cosa ho imparato: fate attenzione alle dimensioni dei vostri file Parquet e alla loro organizzazione.

Inizialmente ho cercato di risolvere la situazione convertendo tutti i TSV in file Parquet. Sono ideali per lavorare con grandi set di dati, perché le informazioni sono memorizzate in formato colonnare: ogni colonna risiede in un proprio segmento di memoria/disk, a differenza dei file di testo, dove le righe contengono gli elementi di ciascuna colonna. E se si deve cercare qualcosa, basta leggere la colonna necessaria. Inoltre, in ogni file, la colonna contiene un intervallo di valori, quindi se il valore cercato non è presente nell'intervallo della colonna, Spark non perderà tempo a esaminare l'intero file.

Ho avviato un'attività semplice AWS Glue per convertire i nostri TSV in Parquet e caricare nuovi file in Athena. Ci è voluto circa 5 ore. Ma quando ho eseguito la query, ci è voluto più o meno lo stesso tempo e un po' meno denaro. Il fatto è che Spark, cercando di ottimizzare il compito, ha semplicemente estratto un chunk TSV e lo ha messo in un chunk Parquet. E poiché ogni chunk era abbastanza grande e conteneva registrazioni complete di molte persone, ogni file conteneva tutti i SNP, quindi Spark doveva aprire tutti i file per estrarre le informazioni necessarie.

È interessante notare che il tipo di compressione predefinito (e raccomandato) in Parquet — snappy — non è suddivisibile (splitable). Pertanto, ogni esecutore (executor) si bloccava nel compito di estrazione e caricamento dell'intero dataset di 3,5 GB.

Analizziamo il problema

Cosa ho imparato: è difficile ordinare, soprattutto se i dati sono distribuiti.

Pensavo di aver finalmente compreso il problema. Dovevo solo ordinare i dati per la colonna SNP, e non per le persone. Allora in un singolo blocco di dati ci sarebbero stati diversi SNP, e allora sarebbe emersa tutta la potenza della funzione 'intelligente' di Parquet di 'aprire solo se il valore è nell'intervallo'. Purtroppo, ordinare miliardi di righe sparse per il cluster si è rivelato un compito complesso.

Me che seguo un corso di algoritmi all'università: «Ugh, a nessuno importa della complessità computazionale di tutti questi algoritmi di ordinamento»

Me che cerco di ordinare in una colonna in un tavolo da 20TB #spark tabella: «Perché ci sta volendo così tanto?» #DataScience difficoltà.

— Nick Strayer (@NicholasStrayer) 11 marzo 2019

AWS decisamente non vuole restituire i soldi per il motivo 'Sono uno studente distratto'. Dopo aver avviato l'ordinamento su Amazon Glue, ha funzionato per 2 giorni e si è concluso con un errore.

E per quanto riguarda il partizionamento?

Cosa ho imparato: le partizioni in Spark devono essere bilanciate.

Poi mi è venuta l'idea di partizionare i dati nei cromosomi. Ce ne sono 23 (e alcuni in più, considerando il DNA mitocondriale e le aree non mappate).
Questo permetterà di dividere i dati in porzioni più piccole. Se aggiungiamo a una funzione Spark di esportazione nel codice Glue solo una riga partition_by = "chr", i dati devono essere suddivisi in bucket.

Il genoma è costituito da numerosi frammenti chiamati cromosomi.

Sfortunatamente, non ha funzionato. I cromosomi hanno dimensioni diverse, ed è quindi presente una quantità di informazioni variabile. Ciò significa che i compiti inviati da Spark ai worker non erano bilanciati e venivano eseguiti lentamente, poiché alcuni nodi terminavano prima e rimanevano inattivi. Tuttavia, le operazioni sono state completate. Ma quando si richiedeva un SNP, l'imbalanzamento ha nuovamente causato problemi. Il costo di elaborazione degli SNP in cromosomi più grandi (cioè da dove vogliamo ottenere i dati) è diminuito solo di circa 10 volte. Molto, ma non abbastanza.

E se dividessimo in partizioni ancora più piccole?

Cosa ho imparato: non cercare mai di creare 2,5 milioni di partizioni.

Ho deciso di andare fino in fondo e ho partizionato ogni SNP. Questo garantiva dimensioni uniformi nelle partizioni. È STATA UNA BRUTTA IDEA. Ho utilizzato Glue e ho aggiunto una semplice riga partition_by = 'snp'. L'attività è stata avviata e ha cominciato a funzionare. Un giorno dopo ho controllato e ho visto che in S3 non era ancora stato registrato nulla, quindi ho interrotto l'attività. Sembra che Glue stesse scrivendo file temporanei in un luogo nascosto in S3, e ci fossero molti file, forse un paio di milioni. Di conseguenza, il mio errore mi è costato più di mille dollari e non ha reso felice il mio mentore.

Partizionamento + ordinamento

Cosa ho imparato: è ancora difficile ordinare, così come configurare Spark.

L'ultimo tentativo di partizionamento prevedeva che partizionassi i cromosomi e poi ordinassi ogni partizione. In teoria, questo avrebbe accelerato ogni richiesta, poiché i dati desiderati sugli SNP avrebbero dovuto trovarsi all'interno di alcuni frammenti Parquet nel range specificato. Purtroppo, l'ordinamento anche dei dati partizionati si è rivelato una sfida ardua. Di conseguenza, sono passato a EMR per un cluster personalizzato e ho utilizzato otto potenti istanze (C5.4xl) e Sparklyr per creare un flusso di lavoro più flessibile...

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

…tuttavia, il compito non è stato comunque completato. Ho provato in diversi modi: aumentando la memoria allocata per ogni esecutore di richieste, utilizzando nodi con più memoria, applicando variabili di broadcasting, ma ogni volta si trattava di misure temporanee, e gradualmente gli esecutori cominciavano a guastarsi, fino a quando tutto si fermava.

Aggiornamento: così inizia. pic.twitter.com/agY4GU2ru5

— Nick Strayer (@NicholasStrayer) 15 maggio 2019

Divento più ingegnoso

Cosa ho imparato: a volte dati particolari richiedono soluzioni particolari.

Ogni SNP ha un valore di posizione. Questo numero corrisponde alla quantità di basi che si trovano lungo il suo cromosoma. È un buon e naturale modo per organizzare i nostri dati. Inizialmente volevo partizionare per aree di ciascun cromosoma. Ad esempio, posizioni 1 – 2000, 2001 – 4000, e così via. Ma il problema è che gli SNP sono distribuiti in modo non uniforme tra i cromosomi, quindi le dimensioni dei gruppi varieranno notevolmente.

Di conseguenza, sono arrivato a una suddivisione per categorie (rank) delle posizioni. Sulla base dei dati già caricati, ho eseguito una query per ottenere un elenco di SNP unici, delle loro posizioni e dei cromosomi. Successivamente, ho ordinato i dati all'interno di ciascun cromosoma e ho raggruppato gli SNP in bin di dimensioni specificate. Ad esempio, 1000 SNP ciascuno. Questo mi ha fornito una correlazione tra SNP e gruppo nel cromosoma.

Alla fine, ho creato gruppi (bin) di 75 SNP, di cui spiegherò il motivo di seguito.

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

Primo tentativo con Spark

Cosa ho imparato: la fusione in Spark funziona rapidamente, ma la partizionamento risulta ancora costoso.

Volevo leggere questo piccolo (2,5 milioni di righe) frame di dati in Spark, unirlo con i dati grezzi e poi partizionare secondo la colonna recentemente aggiunta. 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')
  )

Ho usato sdf_broadcast(), in questo modo Spark capisce che deve inviare il frame dati a tutti i nodi. Questo è utile se i dati sono di dimensioni ridotte e necessari per tutte le attività. Altrimenti, Spark cerca di essere intelligente e distribuisce i dati secondo necessità, il che può causare rallentamenti.

E ancora una volta la mia idea non ha funzionato: i task hanno funzionato per un po', hanno completato la fusione, e poi, come i processi avviati con il partizionamento, hanno cominciato a fallire.

Aggiungo AWK

Cosa ho imparato: non dormite quando vi insegnano le basi. Sicuramente qualcuno ha già risolto il vostro problema negli anni '80.

Fino a quel momento, la causa di tutti i miei fallimenti con Spark era stata la mescolanza dei dati nel cluster. Forse la situazione può migliorare con un preprocessamento. Ho deciso di provare a separare i dati testuali grezzi in colonne cromosomiche, così speravo di fornire a Spark dati "pre-partizionati".

Ho cercato su StackOverflow come dividere per valori delle colonne e ho trovato una risposta così bella. Con AWK, puoi separare un file di testo per valori delle colonne eseguendo un'istruzione nello script, invece di inviare i risultati a stdout.

Per prova, ho scritto uno script Bash. Ho scaricato uno dei pacchetti TSV, poi l'ho estratto usando gzip e inviato a awk.

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

Ha funzionato!

Riempimento dei nuclei

Cosa ho imparato: gnu parallel è una cosa magica, tutti dovrebbero usarla.

La separazione è avvenuta piuttosto lentamente e quando ho avviato htop, per controllare l'utilizzo di un'istanza EC2 potente (e costosa), ho scoperto che stavo utilizzando solo un core e circa 200 MB di memoria. Per risolvere il problema e non perdere molti soldi, dovevo trovare un modo per parallelizzare il lavoro. Fortunatamente, in un libro assolutamente straordinario, Data Science at the Command Line di Jeroen Janssens, ho trovato un capitolo dedicato alla parallelizzazione. Da esso ho appreso di gnu parallel, un metodo molto flessibile per implementare la multithreading in Unix.

Quando ho avviato la separazione utilizzando il nuovo processo, tutto ha funzionato perfettamente, ma c'era un collo di bottiglia: il download degli oggetti S3 sul disco non era troppo veloce e non era completamente parallelizzato. Per risolverlo, ho fatto quanto segue:

1. Ho scoperto che si può implementare direttamente nel pipeline una fase di download S3, escludendo completamente la memorizzazione intermedia su disco. Ciò significa che posso evitare di scrivere i dati grezzi su disco e utilizzare uno spazio di archiviazione AWS ancora più piccolo, e quindi più economico.
2. Con il comando aws configure set default.s3.max_concurrent_requests 50 ho notevolmente aumentato il numero di thread utilizzati da AWS CLI (di default sono 10).
3. Sono passato a un'istanza EC2 ottimizzata per la velocità di rete, con la lettera n nel nome. Ho scoperto che la perdita di potenza di calcolo utilizzando le istanze n è più che ripagata dall'aumento della velocità di caricamento. Per la maggior parte dei compiti ho utilizzato c5n.4xl.
4. Cambiato gzip con pigz, è uno strumento gzip che sa fare cose fantastiche per parallelizzare un'attività di decompressione di file che inizialmente non era parallelizzabile (questo ha aiutato poco).

# 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

Questi passaggi sono combinati tra loro per far funzionare tutto molto velocemente. Grazie all'aumento della velocità di download e all'eliminazione della scrittura su disco, ora potevo elaborare un pacchetto di 5 terabyte in poche ore.

Non c'è niente di più dolce che vedere tutti i core per cui stai pagando su AWS in uso. Grazie a gnu-parallel posso decomprimere e dividere un CSV da 19 gigabyte tanto rapidamente quanto posso scaricarlo. Non riuscivo nemmeno ad avviare spark. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) 17 maggio 2019

Questo tweet doveva menzionare 'TSV'. Purtroppo.

Utilizzo di dati riparserizzati

Cosa ho imparato: Spark ama i dati non compressi e non ama combinare le partizioni.

Ora i dati si trovavano in S3 in un formato non compresso (leggi, condiviso) e semistrutturato, e potevo tornare a Spark. Mi aspettava una sorpresa: non ero riuscito a ottenere ciò che volevo! È stato molto difficile spiegare a Spark come erano partizionati i dati. E anche quando ci sono riuscito, ho scoperto che c'erano troppe partizioni (95.000), e quando ho utilizzato coalesce per ridurle a un numero ragionevole, ho compromesso la mia partizione. Sono certo che si possa risolvere, ma dopo alcuni giorni di ricerca non sono riuscito a trovare una soluzione. Alla fine ho completato tutte le operazioni in Spark, anche se ci è voluto del tempo, e i miei file Parquet divisi non erano molto piccoli (~200 Kb). Tuttavia, i dati erano dove dovevano essere.

Troppo piccoli e diversi, meraviglioso!

Testare le query locali di Spark

Cosa ho imparato: in Spark ci sono troppe spese generali nel risolvere compiti semplici.

Caricando i dati in un formato ben congegnato, sono riuscito a testare la velocità. Ho configurato uno script in R per avviare un server Spark locale, e poi ho caricato un frame di dati Spark dal relativo archivio Parquet. Ho tentato di caricare tutti i dati, ma non sono riuscito a far riconoscere la partizione a Sparklyr.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Inizia il timer
start_time <- Sys.time()

# Carica il bin desiderato in Spark
intensity_data % 
  Spark_read_Parquet(
    name = 'intensity_data', 
    path = get_snp_location(desired_snp),
    memory = FALSE )

# Sottogruppo bin per snp e poi raccogliere a locale
test_subset % 
  filter(SNP_Name == desired_snp) %>% 
  collect()

print(Sys.time() - start_time)

L'esecuzione ha impiegato 29,415 secondi. Molto meglio, ma non sufficientemente veloce per il test di massa di qualcosa. Inoltre, non potevo accelerare l'operazione utilizzando la cache, perché quando provavo a mettere in cache in memoria il frame di dati, Spark si bloccava sempre, anche quando ho allocato più di 50 GB di memoria per un dataset che pesava meno di 15.

Ritorno ad AWK

Cosa ho imparato: gli array associativi in AWK sono molto efficienti.

Ho realizzato che avrei potuto ottenere una velocità maggiore. Mi è tornato in mente che in un fantastico manuale di AWK di Bruce Barnett Ho letto di una funzionalità interessante chiamata «array associativi». In sostanza, si tratta di coppie chiave-valore che in AWK hanno un nome diverso, e quindi non ci ho mai pensato molto. Roman Cheplyaka mi ha ricordato che il termine «array associativi» è molto più antico rispetto al termine «coppia chiave-valore». Anche se cerchi chiave-valore su Google Ngram, non troverai quel termine, ma troverai array associativi! Inoltre, «coppia chiave-valore» è spesso associato ai database, quindi è molto più logico paragonarlo a un hashmap. Ho capito che posso usare questi array associativi per collegare i miei SNP alla tabella dei gruppi (bin table) e ai dati grezzi senza utilizzare Spark.

Per questo, nel mio script AWK ho utilizzato il blocco BEGIN. Questo è un frammento di codice che viene eseguito prima che la prima riga di dati venga passata al corpo principale dello script.

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...) ha caricato tutte le righe dal CSV del gruppo (bin), impostando la prima colonna (nome SNP) come chiave per l'array associativo. bin e il secondo valore (gruppo) come valore. Poi nel blocco { }, che viene eseguito per tutte le righe del file principale, ogni riga viene inviata al file di output, che riceve un nome unico a seconda del suo gruppo (bin): ..._bin_"bin[$1]"_....

Variabili batch_num e chunk_id corrispondevano ai dati forniti dal pipeline, evitando così condizioni di gara, e ogni thread di esecuzione avviato parallelo, scriveva nel proprio file unico.

Poiché avevo distribuito tutti i dati grezzi in cartelle per cromosomi, rimasti dopo il mio precedente esperimento con AWK, ora potevo scrivere un altro script Bash per elaborare un cromosoma alla volta e caricare in S3 dati più approfonditamente partizionati.

DESIRED_CHR='13'

# Scarica i dati del cromosoma da s3 e dividili in 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"

# Combina tutti i chunk dei processi paralleli in file singoli e caricali su rds usando 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/*

Nello script ci sono due sezioni parallelo.

Nella prima sezione, i dati vengono letti da tutti i file contenenti informazioni sul cromosoma richiesto, quindi questi dati vengono distribuiti tra i thread, i quali smistano i file nelle relative categorie (bin). Per evitare stati di competizione, in cui più thread scrivono in un file, AWK passa i nomi dei file per scrivere i dati in luoghi diversi, ad esempio, chr_10_bin_52_batch_2_aa.csv. Di conseguenza, sul disco vengono creati molti piccoli file (per questo ho utilizzato volumi EBS da un terabyte).

Il pipeline della seconda sezione parallelo passa attraverso i gruppi (bin) e unisce i loro file separati in CSV comuni con cat, e poi li invia all'esportazione.

Traslitterazione in R?

Cosa ho imparato: puoi accedere a stdin e stdout dallo script R, il che significa che puoi utilizzarlo nel pipeline.

Nel tuo script Bash, potresti aver notato una riga del tipo: ...cat chunked\/*_bin_{}_*.csv | ./upload_as_rds.R.... Essa trasmette tutti i file concatenati del gruppo (bin) allo script R sottostante. {} è una tecnica speciale parallelo, che inserisce direttamente nel comando stesso tutti i dati inviati in quel flusso specificato. L'opzione {#} fornisce un ID unico del thread di esecuzione, e {%} rappresenta il numero della slot di lavoro (si ripete, ma mai contemporaneamente). Puoi trovare un elenco di tutte le opzioni nella documentazione.

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

Quando la variabile file("stdin") viene passata a readr::read_csv, i dati trasmessi allo script R vengono caricati in un frame, che poi viene salvato come .rds-file utilizzando aws.s3 che scrive direttamente in S3.

RDS è come una versione più semplice di Parquet, senza le complessità di un data store a colonne.

Dopo aver completato lo script Bash, ho ottenuto un insieme di .rds-file presenti in S3, il che mi ha permesso di utilizzare una compressione efficiente e tipi integrati.

Nonostante l'utilizzo di un R poco performante, tutto è andato molto velocemente. Non sorprende che i frammenti in R che gestiscono la lettura e la scrittura dei dati siano ben ottimizzati. Dopo aver testato su un cromosoma di dimensioni medie, il job è stato completato su un'istanza C5n.4xl in circa due ore.

Limiti di S3

Cosa ho imparato: grazie a un'implementazione intelligente dei percorsi, S3 può gestire molti file.

Ero preoccupato che S3 potesse elaborare molti file a essa inviati. Avrei potuto rendere i nomi dei file significativi, ma come farà S3 a cercarli?

Le cartelle in S3 servono solo a scopo estetico, in realtà il sistema non si preoccupa del simbolo. /. Dalla pagina FAQ di S3.

Sembra che S3 rappresenti il percorso per un file specifico come una semplice chiave in una sorta di tabella hash o in un database basato su documenti. Un bucket può essere considerato come una tabella, mentre i file sono le righe di questa tabella.

Poiché velocità ed efficienza sono importanti per generare profitto in Amazon, non sorprende che questo sistema 'chiave-per-punto-di-accesso-al-file' sia estremamente ottimizzato. Ho cercato di trovare un equilibrio: in modo da non dover eseguire numerosi get-request, ma allo stesso tempo mantenere le richieste eseguite rapidamente. È emerso che è meglio avere circa 20.000 file binari. Penso che continuando a ottimizzare si possa migliorare ulteriormente la velocità (ad esempio, creando un bucket specifico solo per i dati, riducendo così la dimensione della tabella di ricerca). Ma non c'era più tempo e denaro per ulteriori esperimenti.

E per quanto riguarda la compatibilità incrociata?

Cosa ho imparato: la principale causa di perdita di tempo è l'ottimizzazione prematura del tuo metodo di archiviazione.

In questo momento è fondamentale chiedersi: «Perché utilizzare un formato di file proprietario?» La risposta sta nella velocità di caricamento (file CSV compressi in gzip caricati sette volte più lentamente) e nella compatibilità con i nostri flussi di lavoro. Posso riconsiderare la mia decisione se R riuscirà a caricare facilmente file Parquet (o Arrow) senza il carico di Spark. Nel nostro laboratorio tutti usano R, e se ho bisogno di convertire i dati in un altro formato, ho ancora a disposizione i dati testuali originali, quindi posso semplicemente riavviare la pipeline.

Divisione del lavoro

Cosa ho imparato: non cercate di ottimizzare i compiti manualmente, lasciate che lo faccia il computer.

Ho ottimizzato il flusso di lavoro su un cromosoma, ora è necessario elaborare tutti gli altri dati.
Volevo avviare diversi istanze EC2 per la conversione, ma allo stesso tempo temevo di ottenere un carico estremamente sbilanciato in diversi compiti di elaborazione (proprio come Spark soffriva per partizioni sbilanciate). Inoltre, non mi andava di avviare un'istanza per ogni cromosoma, perché ci sono limiti predefiniti di 10 istanze per gli account AWS.

Poi ho deciso di scrivere uno script in R per ottimizzare i compiti di elaborazione.

Inizialmente ho chiesto a S3 di calcolare quanto spazio di archiviazione occupa ciascun cromosoma.

library(aws.s3)
library(tidyverse)

chr_sizes % 
  mutate(Size = as.numeric(Size)) %>% 
  filter(Size != 0) %>% 
  mutate(
    # Estrai il cromosoma dal nome del file 
    chr = str_extract(Key, 'chr.{1,4}.csv') %>%
             str_remove_all('chr|.csv')
  ) %>% 
  group_by(chr) %>% 
  summarise(total_size = sum(Size)/1e+9) # Dividi per ottenere il valore 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.
# … con altre 17 righe

Poi ho scritto una funzione che prende la dimensione totale, mescola l'ordine dei cromosomi e li divide in gruppi. num_jobs e comunica quanto variano le dimensioni di tutti i compiti di elaborazione.

num_jobs <- 7
# Quanto sarebbe grande ogni compito se fosse perfettamente diviso?
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.

Poi ho eseguito mille mescolamenti con purrr e ho scelto il migliore.

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

Così ho ottenuto un insieme di compiti molto simili in dimensione. Dopodiché, mi è bastato racchiudere il mio precedente script Bash in un grande ciclo. per. Ci sono voluti circa 10 minuti per scrivere questa ottimizzazione. E questo è decisamente meno di quanto avrei impiegato a creare manualmente i compiti in caso di sbilanciamento. Quindi ritengo che con questa ottimizzazione preliminare, non ho sbagliato.

per DESIRED_CHR in "16" "9" "7" "21" "MT"
do
# Codice per elaborare un singolo cromosoma
fi

Alla fine aggiungo il comando di spegnimento:

sudo shutdown -h now

… ed è andato tutto a buon fine! Con AWS CLI ho avviato le istanze e attraverso l'opzione user_data ho passato loro gli script Bash per l'elaborazione dei compiti. Venivano eseguiti e si spegnevano automaticamente, così non pagavo per capacità computazionale eccessiva.

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

Imballiamo!

Cosa ho imparato: L'API deve essere semplice per semplicità e flessibilità d'uso.

Finalmente ho ottenuto i dati nel posto e nella forma giusti. Dovevo semplificare al massimo il processo di utilizzo dei dati per rendere le cose più facili per i miei colleghi. Volevo creare un'API semplice per effettuare richieste. Se in futuro decidessi di passare a .rds file Parquet, questo dovrebbe essere un problema per me, non per i miei colleghi. Per questo motivo, ho deciso di sviluppare un pacchetto R interno.

Ho assemblato e documentato un pacchetto molto semplice, contenente solo poche funzioni per accedere ai dati, costruito attorno alla funzione get_snp. Ho anche creato un sito per i colleghi su pkgdown, in modo che potessero vedere facilmente esempi e documentazione.

Cache intelligente

Cosa ho imparato: se i vostri dati sono ben preparati, caching sarà facile!

Poiché uno dei principali flussi di lavoro applicava al pacchetto SNP lo stesso modello di analisi, ho deciso di sfruttare il raggruppamento (binning) a mio favore. Quando i dati vengono passati attraverso SNP, tutte le informazioni del gruppo (bin) vengono allegate all'oggetto restituito. In teoria, quindi, le richieste più vecchie possono accelerare il trattamento di nuove richieste.

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

Durante l'assemblaggio del pacchetto, ho eseguito molti benchmark per confrontare la velocità utilizzando diversi metodi. Consiglio di non trascurare questo aspetto, poiché a volte i risultati possono essere inaspettati. Ad esempio, dplyr::filter si è rivelato molto più veloce nella cattura delle righe mediante filtraggio basato su indicizzazione, mentre ottenere una singola colonna dal data frame filtrato funzionava molto più rapidamente rispetto all'uso della sintassi di indicizzazione.

Si noti che l'oggetto prev_snp_results contiene la chiave snps_in_bin. Questo è un array di tutti gli SNP unici nel gruppo (bin), consentendo di controllare rapidamente se ci sono già dati dalla richiesta precedente. Facilita anche il passaggio ciclico attraverso tutti gli SNP nel gruppo (bin) utilizzando questo codice:

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

Risultati

Ora possiamo (e abbiamo iniziato a farlo seriamente) eseguire modelli e scenari che in precedenza non erano a nostra disposizione. La cosa migliore è che i miei colleghi di laboratorio non devono preoccuparsi di complicazioni varie. Hanno semplicemente una funzione funzionante.

E anche se il pacchetto li libera dai dettagli, ho cercato di rendere il formato dei dati abbastanza semplice da poterlo comprendere, nel caso in cui dovessi improvvisamente scomparire domani...

La velocità è aumentata notevolmente. Di solito analizziamo frammenti funzionali significativi del genoma. Prima non potevamo farlo (risultava troppo costoso), ma ora, grazie alla struttura di gruppo (bin) e alla memorizzazione nella cache, per richiedere un SNP ci vogliono in media meno di 0,1 secondi, e l'utilizzo dei dati è così basso che le spese su S3 sono minime.

Recentemente mi hanno incaricato di gestire oltre 25 TB di dati grezzi di genotipizzazione per il mio laboratorio. Quando ho iniziato, usare Spark richiedeva 8 minuti e costava 20 dollari per interrogare un SNP. Dopo aver usato AWK + #rstats per elaborare, ora richiede meno di un decimo di secondo e costa 0,00001 dollari. È una mia personale #BigData vittoria. pic.twitter.com/ANOXVGrmkk

— Nick Strayer (@NicholasStrayer) 30 maggio 2019

Conclusione

Quest'articolo non è affatto una guida. La soluzione trovata è stata personalizzata e quasi sicuramente non ottimale. È piuttosto un racconto di un viaggio. Voglio che gli altri capiscano che simili soluzioni non nascono già formate nella mente, ma sono il risultato di tentativi ed errori. Inoltre, se stai cercando uno specialista in analisi dei dati, tieni presente che per utilizzare efficacemente questi strumenti è necessaria esperienza, e l'esperienza richiede soldi. Sono felice di aver avuto le risorse per pagare, ma molti altri, che potrebbero fare lo stesso lavoro meglio di me, non avranno mai questa possibilità per mancanza di denaro anche solo per provare.

Gli strumenti per i big data sono versatili. Se hai tempo, quasi sicuramente potrai scrivere una soluzione più veloce applicando una «pulizia intelligente» dei dati, un deposito e metodologie di estrazione. In fin dei conti, tutto si riduce all'analisi dei costi e dei benefici.

Cosa ho imparato:

• non esiste un modo economico per analizzare 25 TB in una volta;
• fai attenzione alle dimensioni dei tuoi file Parquet e alla loro organizzazione;
• le partizioni in Spark devono essere bilanciate;
• non cercare mai di creare 2,5 milioni di partizioni;
• ordinare è ancora difficile, così come configurare Spark;
• a volte dati speciali richiedono soluzioni particolari;
• l'unione in Spark è veloce, ma il partizionamento è comunque costoso;
• non dormire quando ti insegnano le basi, qualcuno ha già risolto il tuo problema negli anni '80;
• gnu parallel è qualcosa di magico, tutti dovrebbero usarlo;
• Spark ama i dati non compressi e non ama combinare le partizioni;
• ci sono troppe spese generali in Spark quando si risolvono compiti semplici;
• gli array associativi in AWK sono molto efficienti;
• puoi accedere a stdin e stdout da uno script R, e quindi usarlo nel tuo pipeline;
• grazie a un'implementazione intelligente, i percorsi S3 possono gestire molti file;
• la principale causa di perdita di tempo è l'ottimizzazione prematura del tuo metodo di archiviazione;
• non cercare di ottimizzare i compiti manualmente, lascia che sia il computer a farlo;
• l'API deve essere semplice per garantire facilità e flessibilità d'uso;
• se i tuoi dati sono ben preparati, sarà facile memorizzarli!

Fonte: habr.com