Come leggere questo articolo: mi scuso per il fatto 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 semplicemente la soluzione!» Se volete solo vedere dove sono arrivato, potete passare al capitolo 'Divento più inventivo', ma credo che sia più interessante e utile leggere dei fallimenti.

Recentemente mi è stato assegnato il compito di impostare un processo per analizzare un grande volume di sequenze di DNA (tecnicamente un chip SNP). Dovevo ricevere rapidamente dati su una determinata posizione genetica (chiamata SNP) per la successiva modellazione e altre attività. 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 dimostrerà quale sia stato infine il mio risultato.

Per cominciare, alcune spiegazioni introduttive.

Dati

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

Obiettivo

Come in qualsiasi progetto di gestione dei dati, la cosa più importante era determinare come sarebbero stati utilizzati i dati. In questo caso noi per la maggior parte cercheremo di adattare modelli e flussi di lavoro per SNP basati su SNP. In altre parole, avremo bisogno di dati solo su un singolo SNP alla volta. Dovevo imparare a estrarre nel modo più semplice, rapido ed economico possibile tutte le registrazioni correlate a uno dei 2,5 milioni di SNP.

Come non farlo

Cito un clichè appropriato:

Non ho fallito mille volte, ho solo scoperto mille modi per non analizzare un sacco di dati in un formato utile per le richieste.

Primo tentativo

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

Dopo aver seguito il corso "Metodi avanzati di elaborazione dei Big Data" all'Università di Vanderbilt, ero sicuro che fosse un gioco da ragazzi. Probabilmente ci sarebbero volute un paio d'ore per configurare il server Hive, in modo da elaborare tutti i dati e riportare un risultato. Poiché i nostri dati sono memorizzati in AWS S3, ho utilizzato il servizio Athena, che consente di eseguire query SQL Hive sui dati S3. Non è necessario configurare/avviare un cluster Hive, e si paga solo per i dati cercati.

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

select * from intensityData limit 10;

E ho ricevuto rapidamente risultati ben strutturati. Fatto.

Finché non abbiamo provato a utilizzare i dati in produzione...

Mi è stato chiesto di estrarre tutte le informazioni sui SNP, per testare il modello. Ho eseguito la query:

select * from intensityData 
where snp = 'rs123456';

…e ho cominciato ad aspettare. Dopo otto minuti e più di 4 TB di dati richiesti, ho ricevuto un risultato. Athena addebita in base al volume di dati trovati, a $5 per terabyte. Quindi, questa singola query mi è costata $20 e otto minuti di attesa. Per far funzionare il modello su tutti i dati, avrei dovuto aspettare 38 anni e pagare $50 milioni. È chiaro che non andava bene per noi.

Era necessario utilizzare Parquet…

Cosa ho imparato: fai attenzione alle dimensioni dei tuoi file Parquet e alla loro organizzazione.

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

Ho avviato un compito semplice AWS Glue Per convertire i nostri TSV in Parquet ho caricato nuovi file in Athena. Ci sono volute 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, tentando di ottimizzare il compito, ha semplicemente estratto un chunk di TSV e l'ha messo in un chunk di Parquet. E poiché ogni chunk era piuttosto grande e conteneva record completi di molte persone, in ogni file erano memorizzati tutti i SNP, quindi Spark doveva aprire tutti i file per estrarre le informazioni necessarie.

Curiosamente, il tipo di compressione predefinito (e consigliato) in Parquet - snappy - non è divisibile (splitable). Pertanto, ogni esecutore (executor) rimaneva bloccato durante il compito di estrazione e il caricamento dell'intero dataset di 3,5 GB.

Facciamo chiarezza sulla questione

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

Mi sembrava di aver capito la radice del problema. Avevo solo bisogno di ordinare i dati per la colonna SNP, non per le persone. Allora in un singolo chunk di dati ci sarebbero stati diversi SNP, e allora la funzione 'intelligente' di Parquet di 'aprire solo se il valore è nell'intervallo' si sarebbe manifestata in tutto il suo splendore. Sfortunatamente, ordinare miliardi di righe sparse attraverso il cluster si è rivelato un compito complicato.

Me che prendo un corso di algoritmi al college: «Ugh, a nessuno importa della complessità computazionale di tutti questi algoritmi di ordinamento»

Me mentre cerco di ordinare su una colonna in un 20TB #spark tabella: «Perché ci sta volendo così tanto?» #DataScience fatica.

— Nick Strayer (@NicholasStrayer) 11 marzo 2019

AWS sicuramente non vuole restituire i soldi con la scusa «Sono uno studente distratto». Dopo aver avviato l'ordinamento su Amazon Glue, è durato 2 giorni ed è terminato con un errore.

E per quanto riguarda il partizionamento?

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

Poi mi è venuta in mente l'idea di partizionare i dati per cromosomi. Ce ne sono 23 (e alcuni di più, se consideriamo il DNA mitocondriale e le aree non mappate).
Questo permetterà di suddividere i dati in porzioni più piccole. Se aggiungiamo a Spark la funzione di esportazione nello script Glue solo una riga partition_by = "chr", i dati dovrebbero essere distribuiti in bucket.

Il genoma è composto da numerosi frammenti, chiamati cromosomi.

Sfortunatamente, non ha funzionato. I cromosomi hanno dimensioni diverse, il che significa anche una quantità diversa di informazioni. Questo significa che i compiti che Spark inviava ai lavoratori non erano bilanciati e venivano eseguiti lentamente, perché alcuni nodi terminavano prima e rimanevano inattivi. Tuttavia, i compiti sono stati completati. Ma nel richiedere un SNP, il mancato bilanciamento è tornato a essere motivo di problemi. Il costo di elaborazione di un SNP in cromosomi più grandi (cioè quelli da cui vogliamo ottenere i dati) è diminuito solo di circa 10 volte. Molto, ma non abbastanza.

E se si dividesse in partizioni ancora più piccole?

Cosa ho imparato: in generale, non provate mai a fare 2,5 milioni di partizioni.

Ho deciso di andare fino in fondo e di partizionare ogni SNP. Questo garantiva dimensioni di partizione uguali. È STATA UN'IDEA CATTIVA. Ho utilizzato Glue e aggiunto una riga innocente partition_by = 'snp'. Il compito è stato avviato e ha cominciato a eseguirsi. Un giorno dopo ho controllato e ho visto che in S3 non era ancora stato registrato nulla, quindi ho interrotto il compito. Sembra che Glue scrivesse file intermedi in un luogo nascosto in S3, e molti file, forse un paio di milioni. Di conseguenza, il mio errore è costato più di mille dollari e non ha reso felice il mio mentore.

Partizionamento + ordinamento

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

L'ultimo tentativo di partizionamento consisteva nel partizionare i cromosomi e poi ordinare ogni partizione. In teoria, questo avrebbe accelerato ogni richiesta, perché i dati desiderati sugli SNP avrebbero dovuto trovarsi all'interno di alcuni chunk Parquet all'interno di un dato intervallo. Purtroppo, sortire anche i dati partizionati si è rivelato un compito difficile. 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 tentato in ogni modo: ho aumentato l'allocazione di memoria per ogni esecutore dei compiti, ho utilizzato nodi con più memoria, ho applicato variabili di broadcasting, ma ogni volta si rivelavano misure insufficienti, e progressivamente gli esecutori cominciavano a fallire, fino a fermarsi del tutto.

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

— Nick Strayer (@NicholasStrayer) 15 maggio 2019

Divento più ingegnoso

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

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

Di conseguenza, ho deciso di categorizzare (rank) le posizioni. Sui dati già caricati, ho eseguito una query per ottenere un elenco di SNP unici, le loro posizioni e cromosomi. Poi ho ordinato i dati all'interno di ciascun cromosoma e ho raggruppato gli SNP in gruppi (bin) di dimensione prestabilita. Diciamo, 1000 SNP. Questo mi ha dato la relazione tra SNP e gruppo nel cromosoma.

Alla fine, ho creato gruppi (bin) di 75 SNP, spiegherò il motivo più avanti.

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

Primo tentativo con Spark

Cosa ho imparato: l'unione in Spark funziona rapidamente, ma la partizione continua a costare cara.

Volevo leggere questo piccolo (2,5 milioni di righe) frame di dati in Spark, unire ad esso i dati grezzi e poi partizionare in base alla nuova colonna 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 sa che deve inviare il frame di dati a tutti i nodi. Questo è utile se i dati sono di dimensioni contenute e richiesti per tutte le attività. Altrimenti, Spark cerca di essere intelligente e distribuisce i dati secondo necessità, il che può causare ritardi.

E di nuovo la mia idea non ha funzionato: le attività hanno funzionato per un po', hanno completato l'unione e poi, come i processi avviati dalla partizione, hanno iniziato a guastarsi.

Aggiungo AWK

Cosa ho imparato: non dormire quando ti insegnano le basi. Sicuramente qualcuno ha già risolto il tuo problema negli anni '80.

Fino a questo momento, la causa di tutti i miei fallimenti con Spark era la mescolanza dei dati nel cluster. Forse è possibile migliorare la situazione con una pre-elaborazione. Ho deciso di provare a suddividere i dati di testo grezzo in colonne di cromosomi, così speravo di fornire a Spark dati "pre-partizionati".

Ho cercato su StackOverflow come suddividere in base ai valori delle colonne, e ho trovato una risposta così splendida. Con AWK puoi separare un file di testo in base ai valori delle colonne, registrando uno script, invece di inviare i risultati in stdout.

Per prova ho scritto uno script Bash. Ho scaricato uno dei file TSV compressi e poi l'ho estratto con gzip e inviato in awk.

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

Ha funzionato!

Riempimento dei core

Cosa ho imparato: gnu parallel è una cosa magica che tutti dovrebbero usare.

La separazione procedeva abbastanza lentamente e quando ho avviato htop, per controllare l'uso di un'istanza EC2 potente (e costosa), ho scoperto che utilizzavo solo un core e circa 200 MB di memoria. Per risolvere il problema e non sprecare un sacco di soldi, dovevo trovare un modo per parallelizzare il lavoro. Fortunatamente, in un libro 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 il multithreading in Unix.

Quando ho avviato la separazione con il nuovo processo, andava tutto bene, ma c'era ancora un collo di bottiglia: il download degli oggetti S3 su disco non era particolarmente veloce e non del tutto parallelizzato. Per risolvere questo problema, ho fatto quanto segue:

1. Ho scoperto che potevo implementare direttamente nella pipeline la fase di download da S3, escludendo completamente l'archiviazione temporanea su disco. Ciò significa che posso evitare di scrivere dati grezzi su disco e usare uno storage AWS ancora più piccolo, e quindi più economico.
2. Con il comando aws configure set default.s3.max_concurrent_requests 50 ha aumentato notevolmente il numero di thread utilizzati dall'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 computazionale nell'uso delle istanze n è ampiamente compensata dall'aumento della velocità di caricamento. Per la maggior parte delle attività ho utilizzato c5n.4xl.
4. Ho cambiato gzip in pigz, questo è uno strumento gzip che sa fare cose fantastiche per parallelizzare un'attività di estrazione di file che originariamente non era parallelizzata (questo ha aiutato il meno possibile).

# 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 passi sono stati combinati tra loro affinché tutto funzionasse molto velocemente. Grazie all'aumento della velocità di download e all'abbandono della scrittura su disco, ora potevo elaborare un pacchetto da 5 terabyte in poche ore.

Non c'è niente di più dolce che vedere tutti i core per cui stai pagando su AWS essere utilizzati. Grazie a gnu-parallel, posso estrarre e dividere un CSV da 19 gigabyte tanto velocemente quanto lo posso scaricare. Non riuscivo nemmeno a far funzionare Spark. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) 17 maggio 2019

Questo tweet avrebbe dovuto 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 erano archiviati in S3 in un formato non compresso (leggi, separato) e semi-ordinato, e potevo tornare a Spark. Mi aspettava una sorpresa: non sono riuscito di nuovo a ottenere ciò che volevo! Era molto difficile dire con precisione a Spark come erano partizionati i dati. E anche quando l'ho fatto, si è scoperto che c'erano troppe partizioni (95.000), e quando ho utilizzato coalesce per ridurne il numero a limiti ragionevoli, questo ha rovinato la mia partizionamento. Sono sicuro che si può sistemare, ma dopo un paio di giorni di ricerca non sono riuscito a trovare una soluzione. Alla fine ho completato tutte le attività in Spark, anche se ci è voluto un po' di 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 Spark locali

Cosa ho imparato: in Spark ci sono troppe spese generali nella risoluzione di compiti semplici.

Caricando i dati in un formato pensato, sono riuscito a testare la velocità. Ho impostato uno script in R per avviare un server Spark locale, e poi ho caricato il dataframe Spark dal deposito Parquet specificato (bin). Ho cercato di caricare tutti i dati, ma non sono riuscito a far riconoscere a Sparklyr la partizione.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Iniziare un timer
start_time <- Sys.time()

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

# Sottoinsieme bin a 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 abbastanza bene per testare qualcosa su larga scala. Inoltre, non sono riuscito ad accelerare il lavoro utilizzando la memorizzazione nella cache, perché quando cercavo di memorizzare nella cache il dataframe in memoria, Spark si bloccava sempre, anche quando allocavo 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.

Capivo che potevo ottenere una velocità maggiore. Mi è venuto in mente il fantastico manuale di AWK di Bruce Barnett Ho letto di una funzionalità interessante chiamata «array associativi». In sostanza, sono coppie chiave-valore che in AWK sono state chiamate diversamente, e per questo non mi sono soffermato molto su di esse. Roman Cheplyaka mi ha ricordato che il termine «array associativi» è molto più antico del termine «coppia chiave-valore». Anche se voi cercaste chiave-valore su Google Ngram, non troverete quel termine, ma troverete gli array associativi! Inoltre, la «coppia chiave-valore» è spesso associata ai database, quindi è molto più logico confrontarla con hashmap. Ho capito che posso usare questi array associativi per collegare i miei SNP con la tabella dei gruppi (bin table) e i dati grezzi senza utilizzare Spark.

Per questo, nello script AWK ho usato un 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), ha impostato 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 nel file di output, ottenendo un nome unico a seconda del suo gruppo (bin): ..._bin_"bin[$1]"_....

Variabili batch_num e chunk_id corrispondevano ai dati forniti dalla pipeline, il che ha permesso di evitare condizioni di race, e ogni thread di esecuzione avviato parallel, scriveva nel proprio file unico.

Poiché ho distribuito tutti i dati grezzi in cartelle per cromosomi, rimanendo dal mio precedente esperimento con AWK, ora potevo scrivere un altro script Bash per elaborare un cromosoma alla volta e inviare i dati più profondamente partizionati in S3.

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 in 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 parallel.

Nel primo capitolo vengono letti i dati da tutti i file contenenti informazioni sul cromosoma desiderato, quindi questi dati vengono distribuiti attraverso thread che smistano i file nelle gruppi corrispondenti (bin). Per evitare condizioni di race, quando più thread scrivono su un file, AWK assegna nomi di file per registrare i dati in luoghi diversi, ad esempio, chr_10_bin_52_batch_2_aa.csv. Di conseguenza, sul disco vengono creati numerosi piccoli file (per questo ho utilizzato volumi EBS da un terabyte).

Il pipeline del secondo capitolo parallel attraversa i gruppi (bin) e unisce i loro file separati in un CSV comune cat, e poi li invia per l'esportazione.

Traslazione in R?

Cosa ho imparato: è possibile accedere a stdin e stdout da uno script R, e quindi utilizzarlo nel pipeline.

Nel bash script, potresti aver notato la seguente riga: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Essa trasmette tutti i file concatenati del gruppo (bin) allo script R sottostante. {} è una metodologia speciale parallel, che inserisce direttamente nel comando stesso tutti i dati che invia nel flusso specificato. L'opzione {#} fornisce un ID univoco per il thread di esecuzione, mentre {%} rappresenta il numero dello slot di lavoro (ripetuto, ma mai simultaneamente). Puoi trovare l'elenco di tutte le opzioni in 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 traslati nello script R vengono caricati in un frame, che poi viene salvato come .rds-file e viene registrato direttamente in S3 tramite aws.s3.

RDS è una sorta di versione semplificata di Parquet, senza le complessità dello storage colonnare.

Dopo la conclusione dello script Bash, ho ottenuto un insieme di .rds-file memorizzati in S3, il che mi ha consentito di utilizzare una compressione efficiente e tipi incorporati.

Nonostante l'uso di R lento, tutto funzionava molto velocemente. Non è sorprendente che i frammenti in R responsabili della lettura e scrittura dei dati siano ben ottimizzati. Dopo aver testato su un cromosoma di dimensioni medie, il compito è stato completato su un'istanza C5n.4xl in circa due ore.

Limitazioni di S3

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

Ero preoccupato che S3 potesse gestire un gran numero di file inviati. Avrei potuto rendere i nomi dei file significativi, ma come farà S3 a cercarli?

Le cartelle in S3 sono solo di bellezza, in realtà al sistema non interessa il simbolo /. Dalla pagina FAQ di S3.

Sembra che S3 rappresenti il percorso a un file specifico come una semplice chiave in un particolare hash-table o database basato su documenti. Il bucket può essere considerato come una tabella e i file come record in quella tabella.

Poiché velocità ed efficienza sono importanti per ottenere profitti in Amazon, non sorprende che questo sistema "chiave-per-punto-di-accesso-al-file" sia estremamente ottimizzato. Ho cercato di trovare un equilibrio: non dover effettuare molte richieste get, ma garantire che le richieste venissero eseguite rapidamente. Si è rivelato che era meglio effettuare circa 20.000 file binari. Penso che, continuando a ottimizzare, si possa ottenere un aumento della velocità (ad esempio, creando un bucket speciale solo per i dati, riducendo così la dimensione della tabella di ricerca). Ma non avevo 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 è molto importante chiedersi: "Perché utilizzare un formato di file proprietario?" La risposta risiede nella velocità di caricamento (i file CSV compressi in gzip venivano caricati 7 volte più lentamente) e nella compatibilità con i nostri flussi di lavoro. Posso rivedere la mia decisione se R sarà in grado di caricare facilmente file Parquet (o Arrow) senza il sovraccarico di Spark. Nel nostro laboratorio tutti usano R e se ho bisogno di trasformare i dati in un altro formato, ho comunque i dati testuali originali, quindi posso semplicemente rieseguire la pipeline.

Divisione del lavoro

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

Ho ottimizzato il flusso di lavoro su un cromosoma, ora devo elaborare tutti gli altri dati.
Volevo avviare diverse istanze EC2 per la trasformazione, ma allo stesso tempo temevo di generare un carico estremamente sbilanciato tra i vari compiti di elaborazione (come anche Spark soffriva a causa di partizioni sbilanciate). Inoltre, non mi piaceva l'idea di attivare un'istanza per ogni cromosoma, perché per gli account AWS c'è un limite predefinito di 10 istanze.

Ho quindi deciso di scrivere uno script in R per ottimizzare i compiti di elaborazione.

Prima ho chiesto a S3 di calcolare quanto spazio di archiviazione occupa ogni 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



# Un 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 17 righe aggiuntive

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

num_jobs <- 7
# Quanto sarebbe grande ogni lavoro se perfettamente ripartito?
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)



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

Poi ho eseguito mille mescolamenti usando 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 lavori molto simili per dimensione. Poi restava solo da avvolgere il mio precedente script Bash in un grande ciclo per. Ci sono voluti circa 10 minuti per scrivere questa ottimizzazione. E questo è molto meno di quanto avrei speso per creare manualmente i lavori in caso di squilibrio. Pertanto, ritengo che con questa ottimizzazione preliminare non ho sbagliato.

for 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

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

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

Impacchettiamo!

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

Finalmente ho ottenuto i dati nel posto e nella forma desiderati. Rimaneva da semplificare al massimo il processo di utilizzo dei dati, per facilitare il lavoro ai miei colleghi. Volevo creare un'API semplice per le richieste. Se in futuro decido di passare a .rds Se si tratta di file Parquet, dovrebbe essere un problema per me, non per i colleghi. A tal fine, ho deciso di creare un pacchetto interno in R.

Ho assemblato e documentato un pacchetto molto semplice, che contiene solo alcune funzioni per accedere ai dati, attorno alla funzione get_snp. Ho anche creato un sito per i colleghi pkgdown, in modo che possano facilmente consultare esempi e documentazione.

Caching intelligente

Cosa ho imparato: se i tuoi dati sono ben preparati, il caching sarà semplice!

Poiché uno dei principali flussi di lavoro applicava la stessa modello di analisi al pacchetto SNP, ho deciso di sfruttare il raggruppamento (binning) a mio favore. Quando i dati vengono passati tramite SNP, all'oggetto restituito viene allegata anche tutta l'informazione dal gruppo (bin). Quindi, le richieste precedenti possono (teoricamente) velocizzare l'elaborazione delle 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 la costruzione del pacchetto, ho eseguito numerosi benchmark per confrontare la velocità utilizzando diversi metodi. Consiglio di non ignorare questo aspetto, perché a volte i risultati possono essere inaspettati. Ad esempio, dplyr::filter si è rivelato molto più veloce nel catturare righe tramite filtraggio basato su indicizzazione, e ottenere una colonna da un dataframe filtrato ha funzionato molto più rapidamente rispetto all'uso della sintassi di indicizzazione.

Nota che l'oggetto prev_snp_results contiene la chiave snps_in_bin. Questo è un array di tutti gli SNP unici nel gruppo (bin), che consente di controllare rapidamente se esistono già dati dalla richiesta precedente. Inoltre, semplifica il percorso ciclico tra tutti gli SNP nel gruppo (bin) con 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 sul serio) eseguire modelli e scenari precedentemente inaccessibili. La migliore notizia è che i miei colleghi di laboratorio non devono preoccuparsi di tutte le complessità. Hanno solo una funzione funzionante.

E anche se il pacchetto li solleva dai dettagli, ho cercato di rendere il formato dei dati abbastanza semplice affinché possano capire se domani io dovessi improvvisamente scomparire...

La velocità è notevolmente aumentata. Di solito scansioniamo frammenti funzionalmente significativi del genoma. Prima non potevamo farlo (era troppo costoso), ma ora, grazie alla struttura di raggruppamento (bin) e al caching, una richiesta per uno SNP richiede in media meno di 0,1 secondi, e l'uso dei dati è così basso che le spese per ottenere S3 sono risibili.

Recentemente mi è stato affidato il compito di gestire oltre 25 TB di dati grezzi di genotipizzazione per il mio laboratorio. Quando ho iniziato, utilizzare Spark richiedeva 8 minuti e costava 20$ per interrogare un SNP. Dopo aver utilizzato AWK + #rstats per elaborare, ora richiede meno di un decimo di secondo e costa 0.00001$. Il mio personale #BigData successo. pic.twitter.com/ANOXVGrmkk

— Nick Strayer (@NicholasStrayer) 30 Maggio 2019

Conclusione

Questo articolo non è affatto una guida. La soluzione trovata è stata individuale e quasi sicuramente non ottimale. Piuttosto, è un racconto di un viaggio. Voglio che gli altri capiscano che tali soluzioni non nascono già formate, sono il risultato di prove e errori. Inoltre, se stai cercando un esperto in analisi dei dati, considera che per utilizzare efficacemente questi strumenti serve esperienza, e l'esperienza richiede fondi. Sono felice di avere avuto le risorse per pagare, ma molti altri, che potrebbero fare lo stesso lavoro meglio di me, non avranno mai tale possibilità a causa della mancanza di fondi anche solo per tentare.

Gli strumenti per i big data sono universali. Se hai tempo, quasi sicuramente puoi scrivere una soluzione più veloce, applicando una "intelligente" pulizia dei dati, storage e metodologie di estrazione. In fin dei conti, tutto si riduce ad analizzare costi e benefici.

Cosa ho imparato:

• non esiste un modo economico per analizzare 25 TB tutto in una volta;
• fai attenzione alla dimensione dei tuoi file Parquet e alla loro organizzazione;
• le partizioni in Spark devono essere bilanciate;
• non provare mai a creare 2,5 milioni di partizioni;
• ordinare è ancora difficile, così come configurare Spark;
• a volte dati particolari richiedono soluzioni particolari;
• la fusione in Spark funziona rapidamente, ma la partizione continua a costare caro;
• non dormire quando ti insegnano le basi, sicuramente qualcuno ha già risolto il tuo problema negli anni '80;
• gnu parallel è una cosa magica, tutti dovrebbero usarla;
• Spark ama i dati non compressi e non ama combinare partizioni;
• in Spark ci sono troppe spese generali nella risoluzione di problemi semplici;
• le array associative in AWK sono molto efficaci;
• puoi accedere a stdin e stdout da uno script R, il che significa che puoi usarlo nel pipeline;
• grazie a un'implementazione intelligente dei percorsi S3, può gestire molti file;
• la principale causa di perdita di tempo è l'ottimizzazione prematura del tuo metodo di archiviazione;
• non cercare di ottimizzare i task manualmente, lascia che lo faccia il computer;
• l'API deve essere semplice per semplicità e flessibilità d'uso;
• se i tuoi dati sono ben preparati, sarà facile memorizzare nella cache!

Fonte: habr.com