Wie man diesen Artikel liest: Ich entschuldige mich, dass der Text so lang und chaotisch geworden ist. Um Ihre Zeit zu sparen, beginne ich jedes Kapitel mit einer Einleitung „Was ich gelernt habe“, in der ich in ein bis zwei Sätzen das Wesentliche des Kapitels zusammenfasse.

„Zeig einfach die Lösung!“ Wenn Sie nur sehen möchten, wozu ich gekommen bin, gehen Sie zum Kapitel „Ich werde einfallsreicher“, aber ich finde es interessanter und hilfreicher, über Misserfolge zu lesen.

Kürzlich wurde mir die Aufgabe übertragen, einen Prozess zur Verarbeitung einer großen Menge von DNA-Sequenzen einzurichten (technisch gesehen handelt es sich um ein SNP-Chip). Ich musste schnell Daten zu einem bestimmten genetischen Standort (der SNP genannt wird) für anschließendes Modellieren und weitere Aufgaben erhalten. Mit R und AWK gelang es mir, die Daten auf natürliche Weise zu bereinigen und zu organisieren, wodurch ich die Anfragen erheblich beschleunigte. Das war nicht einfach und erforderte zahlreiche Iterationen. Dieser Artikel hilft Ihnen, einige meiner Fehler zu vermeiden und zeigt, was ich letztendlich erreicht habe.

Zunächst einige einleitende Erklärungen.

Daten

Unser universitäres Zentrum für genetische Informationsverarbeitung stellte uns Daten in Form von TSV mit einem Umfang von 25 TB zur Verfügung. Ich erhielt sie in 5 komprimierten Gzip-Paketen, von denen jedes etwa 240 vier Gigabyte große Dateien enthielt. Jede Zeile enthielt Daten zu einem SNP einer Person. Insgesamt wurden Daten zu ~2,5 Millionen SNPs und ~60.000 Personen übermittelt. Neben den SNP-Informationen gab es in den Dateien zahlreiche Spalten mit Zahlen, die verschiedene Merkmale widerspiegeln, wie z.B. Leseintensität, Häufigkeit verschiedener Allele usw. Insgesamt gab es etwa 30 Spalten mit einzigartigen Werten.

Ziel

Wie in jedem Datenmanagementprojekt war es am wichtigsten zu bestimmen, wie die Daten verwendet werden sollten. In diesem Fall werden wir größtenteils Modelle und Arbeitsabläufe für SNPs basierend auf SNPs anpassen. Das heißt, wir benötigen gleichzeitig Daten nur zu einem SNP. Ich musste lernen, alle Einträge, die sich auf einen der 2,5 Millionen SNPs beziehen, so einfach, schnell und kostengünstig wie möglich zu extrahieren.

So sollte man es nicht machen

Ich zitiere ein passendes Klischee:

Ich habe nicht tausendmal versagt, ich habe nur tausend Wege gefunden, eine Menge Daten in einem für Anfragen geeigneten Format nicht zu parsen.

Erster Versuch

Was ich gelernt habe: Es gibt keinen günstigen Weg, 25 TB auf einmal zu parsen.

Nachdem ich den Kurs "Erweiterte Methoden der Verarbeitung großer Daten" an der Vanderbilt University gehört hatte, war ich mir sicher, dass es ein Kinderspiel wird. Vielleicht würde es ein oder zwei Stunden dauern, den Hive-Server einzurichten, um alle Daten zu durchsuchen und die Ergebnisse zu berichten. Da unsere Daten in AWS S3 gespeichert sind, nutzte ich den Service Athena, der es ermöglicht, Hive SQL-Abfragen auf S3-Daten anzuwenden. Man muss keinen Hive-Cluster einrichten und zahlt nur für die Daten, die man abruft.

Nachdem ich Athena meine Daten und deren Format gezeigt hatte, führte ich einige Tests mit ähnlichen Abfragen durch:

select * from intensityData limit 10;

Und erhielt schnell gut strukturierte Ergebnisse. Fertig.

Bis wir versuchten, die Daten in der Praxis zu verwenden…

Ich wurde gebeten, alle Informationen zu SNPs abzurufen, um sie in einem Modell zu testen. Ich führte die Abfrage aus:

select * from intensityData 
where snp = 'rs123456';

…und wartete. Nach acht Minuten und mehr als 4 TB abgerufener Daten erhielt ich das Ergebnis. Athena berechnet Gebühren für das Volumen der gefundenen Daten, 5 $ pro Terabyte. Also kostete diese einzige Abfrage 20 $ und acht Minuten Wartezeit. Um das Modell über alle Daten zu laufen zu lassen, müsste man 38 Jahre warten und 50 Millionen $ zahlen. Das war offensichtlich keine Option für uns.

Es war notwendig, Parquet zu verwenden…

Was ich gelernt habe: Achten Sie auf die Größe Ihrer Parquet-Dateien und deren Organisation.

Zuerst versuchte ich, die Situation zu verbessern, indem ich alle TSV in Parquet-Dateienkonvertierte. Sie sind ideal für die Arbeit mit großen Datensätzen, weil die Informationen in kolumnarer Form gespeichert werden: Jede Spalte befindet sich in ihrem eigenen Speichersegment, im Gegensatz zu Textdateien, in denen die Zeilen Elemente jeder Spalte enthalten. Wenn etwas gefunden werden muss, reicht es aus, die erforderliche Spalte zu lesen. Außerdem speichert jede Datei in der Spalte einen Wertebereich, sodass, wenn das gesuchte Wert nicht im Bereich der Spalte enthalten ist, Spark keine Zeit mit dem Scannen der gesamten Datei verschwendet.

Ich startete eine einfache Aufgabe AWS Glue Um unsere TSV in Parquet zu konvertieren, habe ich neue Dateien in Athena hochgeladen. Das dauerte etwa 5 Stunden. Aber als ich die Abfrage startete, brauchte sie ungefähr die gleiche Zeit und kostete ein wenig weniger Geld. Das liegt daran, dass Spark, um die Aufgabe zu optimieren, einfach einen TSV-Chunk entpackte und ihn in einen eigenen Parquet-Chunk legte. Und da jeder Chunk ziemlich groß war und vollständige Datensätze vieler Personen enthielt, waren in jeder Datei alle SNPs gespeichert, weshalb Spark alle Dateien öffnen musste, um die benötigten Informationen zu extrahieren.

Interessanterweise ist der standardmäßig (und empfohlen) verwendete Kompressionstyp in Parquet - Snappy - nicht teilbar (splitable). Daher hing jeder Executor bei der Entpackung und dem Hochladen des vollständigen Datensatzes von 3,5 GB fest.

Wir klären das Problem.

Was ich gelernt habe: Sortieren ist schwierig, besonders wenn die Daten verteilt sind.

Ich dachte, ich hätte das Problem jetzt verstanden. Ich musste die Daten nur nach der SNP-Spalte und nicht nach den Personen sortieren. Dann würden in einem einzelnen Daten-Chunk mehrere SNPs gespeichert, und die „intelligente“ Funktion von Parquet, „nur zu öffnen, wenn der Wert im Bereich liegt“, könnte sich optimal entfalten. Leider erwies es sich als schwierig, Milliarden von Zeilen, die über den Cluster verteilt sind, zu sortieren.

Ich, der an der Universität einen Algorithmuskurs belegt: „Ugh, niemand interessiert sich für die Berechnungskomplexität all dieser Sortieralgorithmen“

Ich, der versuche, in einer 20TB-Tabelle nach einer Spalte zu sortieren: „Warum dauert das so lange?“ #spark Kämpfe. #DataScience — Nick Strayer (@NicholasStrayer)

11. März 2019 AWS möchte auf keinen Fall Geld zurückgeben mit der Begründung „Ich bin ein zerstreuter Student“. Nachdem ich die Sortierung in Amazon Glue gestartet hatte, lief sie 2 Tage lang und schlug fehl.

Was ist mit der Partitionierung?

: Die Partitionen in Spark sollten ausgewogen sein.

Was ich gelernt habeDann kam mir die Idee, die Daten nach Chromosomen zu partitionieren. Es gibt 23 Stück (und noch einige mehr, wenn man die mitochondriale DNA und nicht entschlüsselte (unmapped) Bereiche berücksichtigt).

Das ermöglicht es, die Daten in kleinere Portionen zu teilen. Wenn ich einfach eine Zeile in die Spark-Exportfunktion im Glue-Skript hinzufügen würde
partition_by = "chr" , dann sollten die Daten in Buckets verteilt werden.Das Genom besteht aus zahlreichen Fragmenten, die Chromosomen genannt werden.

Das Genom besteht aus zahlreichen Fragmenten, die Chromosomen genannt werden.

Leider hat das nicht funktioniert. Chromosomen haben unterschiedliche Größen, was bedeutet, dass sie auch verschiedene Mengen an Informationen enthalten. Das bedeutet, dass die Aufgaben, die Spark an die Worker gesendet hat, nicht ausgeglichen waren und langsam ausgeführt wurden, weil einige Knoten früher fertig waren und untätig waren. Die Aufgaben wurden jedoch ausgeführt. Aber bei der Abfrage eines SNP führte das Ungleichgewicht erneut zu Problemen. Die Kosten für die Verarbeitung von SNP in größeren Chromosomen (also dort, wo wir die Daten erhalten möchten) haben sich nur um etwa das Zehnfache reduziert. Viel, aber nicht genug.

Was, wenn wir in noch kleinere Partitionen aufteilen?

Was ich gelernt habe: Versuchen Sie niemals, 2,5 Millionen Partitionen zu erstellen.

Ich habe beschlossen, es ganz auszureizen und jedes SNP zu partitionieren. Das garantierte eine einheitliche Größe der Partitionen. DAS WAR EINE SCHLECHTE IDEE. Ich habe Glue verwendet und eine harmlose Zeile hinzugefügt partition_by = 'snp'. Die Aufgabe wurde gestartet und begann zu laufen. Einen Tag später habe ich nachgesehen und festgestellt, dass in S3 immer noch nichts gespeichert war, also habe ich die Aufgabe abgebrochen. Es scheint, dass Glue Zwischen-Dateien an einem versteckten Ort in S3 gespeichert hat und viele Dateien, möglicherweise ein paar Millionen. Das Resultat war, dass mein Fehler mehr als tausend Dollar gekostet hat und meinen Mentor nicht erfreute.

Partitionierung + Sortierung

Was ich gelernt habe: Sortieren ist immer noch eine Herausforderung, ebenso wie die Einrichtung von Spark.

Der letzte Versuch der Partitionierung bestand darin, die Chromosomen zu partitionieren und dann jede Partition zu sortieren. Theoretisch würde das jede Anfrage beschleunigen, da die gewünschten SNP-Daten innerhalb weniger Parquet-Chunk im angegebenen Bereich liegen sollten. Leider stellte sich die Sortierung selbst partitionierter Daten als schwierige Aufgabe heraus. Daher wechselte ich zu EMR für einen benutzerdefinierten Cluster und nutzte acht leistungsstarke Instanzen (C5.4xl) und Sparklyr, um einen flexibleren Workflow zu schaffen…

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

…trotzdem wurde die Aufgabe nicht abgeschlossen. Ich habe alles Mögliche eingestellt: die Speicherausstattung für jeden Query-Executor erhöht, Knoten mit mehr Speicher verwendet, Broadcast-Variablen angewandt, aber jedes Mal blieb es bei Halbherzigkeiten, und allmählich begannen die Executor zu fehlern, bis alles zum Stillstand kam.

Update: Und es beginnt. pic.twitter.com/agY4GU2ru5

11. März 2019 15. Mai 2019

Ich werde einfallsreicher

Was ich gelernt habe: Manchmal erfordern besondere Daten spezielle Lösungen.

Jeder SNP hat einen Positionswert. Diese Zahl entspricht der Anzahl der Basen, die entlang seines Chromosoms liegen. Das ist eine gute und natürliche Möglichkeit, unsere Daten zu organisieren. Zuerst wollte ich nach Bereichen jedes Chromosoms partitionieren. Zum Beispiel die Positionen 1 — 2000, 2001 — 4000 usw. Aber das Problem ist, dass SNPs ungleichmäßig über die Chromosomen verteilt sind, daher werden die Gruppengrößen stark variieren.

Als Ergebnis kam ich zu einer kategorisierten (Rank) Aufteilung der Positionen. Für die bereits geladenen Daten habe ich eine Abfrage durchgeführt, um eine Liste einzigartiger SNPs, deren Positionen und Chromosomen zu erhalten. Dann sortierte ich die Daten innerhalb jedes Chromosoms und gruppierte die SNPs in Gruppen (Bins) fester Größe. Sagen wir, jeweils 1000 SNPs. Das gab mir einen Zusammenhang zwischen SNP und Gruppe-im-Chromosom.

Letztendlich habe ich Gruppen (Bins) von 75 SNPs erstellt, den Grund erkläre ich unten.

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

Der erste Versuch mit Spark

Was ich gelernt habe: Das Zusammenführen in Spark funktioniert schnell, aber die Partitionierung ist immer noch teuer.

Ich wollte diesen kleinen (2,5 Millionen Zeilen) DataFrame in Spark lesen, ihn mit Rohdaten zusammenführen und dann nach der neu hinzugefügten Spalte partitionieren. 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')
  )

Ich habe sdf_broadcast(), damit Spark weiß, dass es den DataFrame an alle Knoten senden muss. Das ist nützlich, wenn die Daten klein sind und für alle Aufgaben benötigt werden. Andernfalls versucht Spark, smart zu sein und verteilt die Daten nach Bedarf, was zu Verzögerungen führen kann.

Und wieder hat meine Idee nicht funktioniert: Die Aufgaben liefen eine Weile, beendeten das Zusammenführen und begannen dann, wie die gestarteten Partitionierungs-Executoren, zu fehlschlagen.

Ich füge AWK hinzu

Was ich gelernt habe: Schlaft nicht, wenn euch die Grundlagen beigebracht werden. Sicherlich hat jemand euer Problem schon in den 1980er Jahren gelöst.

Bis zu diesem Zeitpunkt war der Grund für all meine Misserfolge mit Spark die Vermischung der Daten im Cluster. Vielleicht kann die Situation durch Vorverarbeitung verbessert werden. Ich entschloss mich, die Rohtextdaten in Chromosomenspalten zu unterteilen, damit ich Spark "vorher partitionierte" Daten anbieten kann.

Ich suchte auf StackOverflow danach, wie man nach Werten in Spalten aufteilen kann, und fand eine so wunderbare Antwort. Mit AWK können Sie eine Textdatei basierend auf den Spaltenwerten aufteilen, indem Sie dies in einem Skript aufzeichnen und die Ergebnisse nicht an stdout.

Zum Testen habe ich ein Bash-Skript geschrieben. Ich habe eine der gepackten TSV-Dateien heruntergeladen und sie anschließend mit gzip entspackt und an awk.

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

Das hat funktioniert!

Auflagen von Kernen

Was ich gelernt habe: gnu parallel ist eine wundersame Sache, die jeder nutzen sollte.

Das Teilen verlief ziemlich langsam, und als ich htop, um die Auslastung eines leistungsstarken (und teuren) EC2-Instances zu überprüfen, stellte ich fest, dass ich nur einen Kern und etwa 200 MB Speicher verwendete. Um dieses Problem zu lösen und nicht viel Geld zu verlieren, musste ich mir überlegen, wie ich die Arbeit parallelisieren kann. Glücklicherweise fand ich in dem absolut erstaunlichen Buch Data Science at the Command Line von Jeroen Janssens ein Kapitel über Parallelisierung. Dort erfuhr ich von gnu parallel, einer sehr flexiblen Methode zur Implementierung von Multithreading in Unix.

Als ich die Teilung mit dem neuen Prozess startete, lief alles wunderbar, aber es gab einen Engpass – das Herunterladen von S3-Objekten auf die Festplatte war nicht besonders schnell und nicht vollständig parallelisiert. Um dies zu beheben, tat ich Folgendes:

1. Ich stellte fest, dass ich den S3-Download-Schritt direkt in der Pipeline implementieren kann, wodurch die Zwischenablage auf der Festplatte vollständig ausgeschlossen wird. Das bedeutet, dass ich die Rohdaten nicht auf der Festplatte speichern und ein noch kleineres und damit günstigeres Speicher-Setup bei AWS verwenden kann.
2. Befehl aws configure set default.s3.max_concurrent_requests 50 erhöhte stark die Anzahl der Threads, die AWS CLI verwendet (standardmäßig sind es 10).
3. Ich wechselte zu einer netzwerkoptimierten EC2-Instanz mit dem Buchstaben n im Namen. Ich stellte fest, dass der Verlust an Rechenleistung bei der Verwendung von n-Instances mehr als durch die erhöhte Upload-Geschwindigkeit ausgeglichen wird. Für die meisten Aufgaben verwendete ich c5n.4xl.
4. Ich änderte gzip auf pigz, dies ist ein gzip-Werkzeug, das großartige Dinge für die Parallelisierung ursprünglich nicht parallelisierter Aufgaben zur Entpackung von Dateien leisten kann (das half am wenigsten).

# 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

Diese Schritte wurden kombiniert, um alles sehr schnell arbeiten zu lassen. Durch die erhöhte Download-Geschwindigkeit und die Vermeidung der Speicherung auf der Festplatte konnte ich jetzt ein 5-Terabyte-Paket in nur wenigen Stunden bearbeiten.

Nichts ist süßer, als alle Kerne zu sehen, für die man bei AWS bezahlt, die auch wirklich genutzt werden. Dank gnu-parallel kann ich eine 19 GB große CSV-Datei genauso schnell entpacken und aufteilen, wie ich sie herunterladen kann. Ich konnte nicht einmal Spark zum Laufen bringen. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

11. März 2019 17. Mai 2019

In diesem Tweet hätte 'TSV' erwähnt werden sollen. Leider.

Wiederverwendung neu parsierter Daten

Was ich gelernt habe: Spark liebt unkomprimierte Daten und mag es nicht, Partitionen zu kombinieren.

Jetzt lagen die Daten in S3 in einem unkomprimierten (d.h. durchsuchbaren) und halbgeordneten Format, und ich konnte zu Spark zurückkehren. Ich wurde überrascht: Ich konnte wieder nicht das gewünschte Ergebnis erzielen! Es war sehr schwierig, Spark genau zu sagen, wie die Daten partitioniert sind. Und selbst als ich es tat, stellte sich heraus, dass es zu viele Partitionen gab (95.000), und als ich mithilfe von coalesce ihre Anzahl auf vernünftige Grenzen reduzierte, wurde meine Partitionierung zunichtegemacht. Ich bin sicher, das lässt sich beheben, aber nach ein paar Tagen Suche konnte ich keine Lösung finden. Schließlich schloss ich alle Aufgaben in Spark ab, obwohl es eine Weile dauerte, und meine aufgeteilten Parquet-Dateien waren nicht sehr klein (~200 KB). Aber die Daten waren genau dort, wo sie sein sollten.

Zu klein und ungleich, wunderbar!

Testen von lokalen Spark-Abfragen

Was ich gelernt habe: In Spark gibt es zu viel Overhead bei der Lösung einfacher Aufgaben.

Nachdem ich die Daten in einem durchdachten Format hochgeladen hatte, konnte ich die Geschwindigkeit testen. Ich richtete ein Skript in R ein, um einen lokalen Spark-Server zu starten, und lud dann den Spark-Datenrahmen aus dem angegebenen Parquet-Gruppen-Repository (bin) hoch. Ich versuchte, alle Daten zu laden, konnte Sparklyr jedoch nicht dazu bringen, die Partitionierung zu erkennen.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Timer starten
start_time <- Sys.time()

# Die gewünschte Binärdatei in Spark laden
intensity_data %
  Spark_read_Parquet(
    name = 'intensity_data', 
    path = get_snp_location(desired_snp),
    memory = FALSE )

# Binärdatei nach SNP unterteilen und dann lokal sammeln
test_subset %
  filter(SNP_Name == desired_snp) %>%
  collect()

print(Sys.time() - start_time)

Die Ausführung dauerte 29,415 Sekunden. Viel besser, aber nicht gut genug für massives Testen von irgendetwas. Außerdem konnte ich die Leistung nicht durch Caching beschleunigen, da Spark immer abstürzte, wenn ich versuchte, den Datenrahmen im Speicher zu cachen, selbst als ich mehr als 50 GB Speicher für einen Datensatz zugewiesen hatte, der weniger als 15 wiegte.

Rückkehr zu AWK

Was ich gelernt habe: assoziative Arrays in AWK sind sehr effizient.

Ich wusste, dass ich eine höhere Geschwindigkeit erzielen kann. Es erinnerte mich an das wunderbare Handbuch zu AWK von Bruce Barnett Ich habe von einem coolen Feature gelesen, das "assoziative Arrays" genannt wird. Tatsächlich sind das Schlüssel-Wert-Paare, die in AWK aus irgendeinem Grund anders genannt werden, weshalb ich sie nicht wirklich im Kopf hatte. Roman Cheplyaka hat mich daran erinnert, dass der Begriff "assoziative Arrays" viel älter ist als der Begriff "Schlüssel-Wert-Paar". Selbst wenn Sie Schlüssel-Wert-Paar bei Google Ngram, werden Sie diesen Begriff dort nicht sehen, stattdessen finden Sie assoziative Arrays! Außerdem wird "Schlüssel-Wert-Paar" meist mit Datenbanken assoziiert, weshalb es viel logischer ist, mit Hashmaps zu vergleichen. Ich habe verstanden, dass ich diese assoziativen Arrays nutzen kann, um meine SNP mit der Bin-Tabelle (bin table) und Rohdaten ohne Verwendung von Spark zu verbinden.

Dafür habe ich im AWK-Skript den Block BEGIN. Dies ist ein Codeabschnitt, der ausgeführt wird, bevor die erste Zeile der Daten an den Hauptteil des Skripts übergeben wird.

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...) hat alle Zeilen aus der CSV-Gruppe (bin) geladen, die erste Spalte (SNP-Name) als Schlüssel für das assoziative Array gesetzt bin und das zweite Wert (Gruppe) als Wert. Dann wird im Block { }, der für alle Zeilen der Hauptdatei ausgeführt wird, jede Zeile in die Ausgabedatei gesendet, die einen einzigartigen Namen je nach ihrer Gruppe (bin) erhält: ..._bin_"bin[$1]"_....

Variablen batch_num und chunk_id entsprachen den vom Pipeline bereitgestellten Daten, was es ermöglichte, Zustände von Rennen zu vermeiden, und jeder Ausführungsstrang, der gestartet wurde parallel, schrieb in seine eigene, einzigartige Datei.

Da ich alle Rohdaten nach Chromosomen in Ordner verteilt hatte, die aus meinem vorherigen Experiment mit AWK übrig geblieben waren, konnte ich jetzt ein anderes Bash-Skript schreiben, um chromosomweise zu verarbeiten und tiefer partitionierte Daten in S3 zurückzugeben.

DESIRED_CHR='13'

# Download Chromosomen-Daten von s3 und in Bins aufteilen
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"

# Kombinieren Sie alle parallelen Prozessabschnitte zu einzelnen Dateien und laden Sie sie mit R in RDS hoch
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/*

Im Skript gibt es zwei Abschnitte parallel.

Im ersten Abschnitt werden die Daten aus allen Dateien, die Informationen über das benötigte Chromosom enthalten, eingelesen. Anschließend werden diese Daten auf Threads verteilt, die die Dateien auf die entsprechenden Gruppen (Bin) aufteilen. Um Bedingungen für einen Wettlauf zu vermeiden, wenn mehrere Threads in eine Datei schreiben, übergibt AWK die Dateinamen für das Schreiben der Daten an verschiedene Orte, zum Beispiel chr_10_bin_52_batch_2_aa.csv. Infolgedessen entsteht auf der Festplatte eine Vielzahl kleiner Dateien (dazu habe ich Terabyte EBS-Volumes verwendet).

Die Pipeline im zweiten Abschnitt parallel geht durch die Gruppen (Bin) und kombiniert deren separaten Dateien in gemeinsame CSVs mit cat, und sendet sie anschließend zum Export.

über R streamen?

Was ich gelernt habe: man kann darauf aus stdin und stdout einem R-Skript zugreifen, und somit kann es auch in der Pipeline verwendet werden.

Im Bash-Skript haben Sie möglicherweise folgende Zeile bemerkt: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Sie streamt alle verketteten Dateien der Gruppe (Bin) in das nachfolgende R-Skript. {} ist eine spezielle Methode, parallel, die alle von ihr an den angegebenen Stream gesendeten Daten direkt in den Befehl einfügt. Die Option {#} bietet eine eindeutige ID des Verarbeitungsthreads, und {%} stellt die Slotnummer der Aufgabe dar (wiederholt sich, aber niemals gleichzeitig). Eine Liste aller Optionen finden Sie in der 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
)

Wenn die Variable file("stdin") an readr::read_csv, werden die an das R-Skript streamten Daten in einen Frame geladen, der dann als .rds-Datei mithilfe von aws.s3 direkt in S3 geschrieben.

RDS ist eine Art vereinfachte Version von Parquet, ohne die Raffinessen eines spaltenorientierten Speichers.

Nach Abschluss des Bash-Skripts erhielt ich eine Reihe von .rds-Dateien, die in S3 liegen, was mir ermöglichte, effiziente Kompression und integrierte Datentypen zu verwenden.

Trotz der Verwendung des langsamen R funktionierte alles sehr schnell. Es ist nicht überraschend, dass die Teile in R, die für das Lesen und Schreiben von Daten verantwortlich sind, gut optimiert sind. Nach Tests mit einem mittelgroßen Chromosom wurde die Aufgabe auf einer C5n.4xl-Instanz in etwa zwei Stunden fertiggestellt.

S3-Beschränkungen

Was ich gelernt habe: Dank einer cleveren Implementierung der S3-Pfade kann S3 viele Dateien verarbeiten.

Ich war besorgt, ob S3 die vielen ihr übergebenen Dateien bewältigen könnte. Ich könnte die Dateinamen sinnvoll gestalten, aber wie würde S3 danach suchen?

Die Ordner in S3 sind nur zur Dekoration, tatsächlich interessiert sich das System nicht für das Symbol /. Von der FAQ-Seite von S3.

Es scheint, dass S3 den Pfad zu einer bestimmten Datei als einfachen Schlüssel in einer Art Hash-Tabelle oder dokumentenbasierter Datenbank darstellt. Ein Bucket kann als Tabelle betrachtet werden, während die Dateien als Datensätze in dieser Tabelle fungieren.

Da Geschwindigkeit und Effizienz wichtig für die Rentabilität bei Amazon sind, ist es kaum überraschend, dass dieses „Schlüssel-zum-Pfad-zur-Datei“-System extrem optimiert ist. Ich habe versucht, ein Gleichgewicht zu finden: um nicht viele Get-Anfragen machen zu müssen, aber damit die Anfragen trotzdem schnell ausgeführt werden. Es stellte sich heraus, dass es am besten ist, etwa 20.000 Bin-Dateien zu erstellen. Ich denke, wenn wir weiter optimieren, könnte die Geschwindigkeit gesteigert werden (zum Beispiel indem wir einen speziellen Bucket nur für Daten erstellen, wodurch die Größe der Suchdatenbank verringert wird). Aber für weitere Experimente fehlte mir bereits die Zeit und das Geld.

Was ist mit der Interoperabilität?

Was ich gelernt habe: Der Hauptgrund für Zeitverschwendung ist die vorzeitige Optimierung deiner Speicherstrategie.

In diesem Moment ist es sehr wichtig, sich zu fragen: „Warum ein proprietäres Dateiformat verwenden?“ Der Grund liegt in der Ladegeschwindigkeit (komprimierte Gzip-CSV-Dateien luden sich sieben Mal langsamer) und der Kompatibilität mit unseren Arbeitsabläufen. Ich könnte meine Entscheidung überdenken, wenn R in der Lage ist, Parquet- (oder Arrow-)Dateien problemlos ohne die Last von Spark zu laden. In unserem Labor nutzen alle R, und wenn ich die Daten in ein anderes Format umwandeln muss, habe ich immer noch die ursprünglichen Textdaten, sodass ich die Pipeline einfach erneut starten kann.

Arbeitsteilung

Was ich gelernt habe: Versuche nicht, Aufgaben manuell zu optimieren, lass das den Computer erledigen.

Ich habe den Workflow an einem Chromosom optimiert, jetzt müssen die restlichen Daten verarbeitet werden.
Ich wollte mehrere EC2-Instanzen zum Transformieren hochheben, hatte aber auch Bedenken, eine stark unausgewogene Last in den verschiedenen Verarbeitungsaufgaben zu erhalten (so wie Spark unter unausgewogenen Partitionen litt). Außerdem wollte ich nicht für jedes Chromosom eine Instanz hochheben, da es eine Standardbeschränkung von 10 Instanzen für AWS-Konten gibt.

Dann beschloss ich, ein Skript in R zu schreiben, um die Verarbeitungsaufgaben zu optimieren.

Zunächst bat ich S3 zu berechnen, wie viel Speicherplatz jedes Chromosom benötigt.

library(aws.s3)
library(tidyverse)

chr_sizes % 
  mutate(Size = as.numeric(Size)) %>% 
  filter(Size != 0) %>% 
  mutate(
    # Extrahiere das Chromosom aus dem Dateinamen 
    chr = str_extract(Key, 'chr.{1,4}.csv') %>%
             str_remove_all('chr|.csv')
  ) %>% 
  group_by(chr) %>% 
  summarise(total_size = sum(Size)/1e+9) # Teilen, um den Wert in GB zu erhalten



# Ein 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.
# … mit 17 weiteren Zeilen

Dann habe ich eine Funktion geschrieben, die die Gesamtdimension nimmt, die Chromosomenreihenfolge mischt und sie in Gruppen einteilt. num_jobs und meldet, wie unterschiedlich die Größen aller Bearbeitungsjobs sind.

num_jobs <- 7
# Wie groß wäre jeder Job, wenn er perfekt aufgeteilt wäre?
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)



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

Dann habe ich mit purrr tausend Mischungen durchgeführt und die beste ausgewählt.

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

So habe ich einen Satz von Jobs erhalten, die in der Größe sehr ähnlich sind. Dann musste ich nur noch mein vorheriges Bash-Skript in eine große Schleife einpacken. for. Für das Schreiben dieser Optimierung habe ich etwa 10 Minuten gebraucht. Und das ist viel weniger, als ich mit der manuellen Erstellung von Jobs bei deren Ungleichgewicht aufgewendet hätte. Deshalb denke ich, dass ich mit dieser vorläufigen Optimierung nicht falsch lag.

for DESIRED_CHR in "16" "9" "7" "21" "MT"
do
# Code zur Verarbeitung eines einzelnen Chromosoms
fi

Am Ende füge ich den Ausschaltbefehl hinzu:

sudo shutdown -h now

… und es hat geklappt! Mit dem AWS CLI habe ich Instanzen hochgefahren und über die Option user_data habe ich ihnen meine Bash-Skripte für die Bearbeitungsjobs übergeben. Diese wurden ausgeführt und schalteten sich automatisch aus, sodass ich nicht für übermäßige Rechenleistung zahlen musste.

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

Packen wir es ein!

Was ich gelernt habe: Die API sollte einfach sein, um Einfachheit und Flexibilität bei der Nutzung zu gewährleisten.

Endlich habe ich die Daten am richtigen Ort und in der gewünschten Form. Es blieb nur noch, den Prozess der Datennutzung so weit wie möglich zu vereinfachen, damit es meinen Kollegen leichter fällt. Ich wollte eine einfache API für die Erstellung von Anfragen machen. Wenn ich mich in Zukunft entscheide, zu wechseln mit .rds Bei Parquet-Dateien sollte dies ein Problem für mich sein und nicht für meine Kollegen. Deshalb habe ich beschlossen, ein internes R-Paket zu erstellen.

Ich habe ein sehr einfaches Paket zusammengestellt und dokumentiert, das nur ein paar Funktionen für den Datenzugriff enthält, die sich um die Funktion get_snpdrehen. Außerdem habe ich für meine Kollegen eine Website erstellt pkgdown, damit sie Beispiele und Dokumentation einfach einsehen können.

Intelligentes Caching

Was ich gelernt habe: Wenn Ihre Daten gut vorbereitet sind, wird das Caching einfach!

Da einer der Hauptarbeitsabläufe bei dem SNP-Paket dasselbe Analysemodell anwendete, habe ich beschlossen, das Gruppieren (Binning) zu meinem Vorteil zu nutzen. Bei der Übertragung von Daten über SNP wird auch alle Informationen aus der Gruppe (Bin) an das zurückgegebene Objekt angehängt. Das bedeutet, dass alte Anfragen (theoretisch) die Verarbeitung neuer Anfragen beschleunigen können.

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

Beim Erstellen des Pakets habe ich viele Benchmarks durchgeführt, um die Geschwindigkeit unter verschiedenen Methoden zu vergleichen. Ich empfehle, dies nicht zu vernachlässigen, da die Ergebnissen manchmal überraschend sind. Zum Beispiel dplyr::filter war viel schneller beim Erfassen von Zeilen mithilfe von indexbasiertem Filtern, und das Abrufen einer Spalte aus dem gefilterten Datenrahmen funktionierte viel schneller als die Verwendung der Indexierungssyntax.

Bitte beachten Sie, dass das Objekt prev_snp_results den Schlüssel snps_in_binenthält. Dies ist ein Array aller einzigartigen SNPs in der Gruppe (Bin), das eine schnelle Überprüfung ermöglicht, ob bereits Daten aus der vorherigen Anfrage vorhanden sind. Es vereinfacht auch das zyklische Durchlaufen aller SNPs in der Gruppe (Bin) mit diesem Code:

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

Ergebnisse

Jetzt können wir (und haben ernsthaft begonnen) Modelle und Szenarien durchzuführen, die uns zuvor nicht möglich waren. Das Beste daran ist, dass meine Kollegen im Labor sich keine Gedanken über komplizierte Details machen müssen. Sie haben einfach eine funktionierende Funktion.

Und obwohl das Paket sie von den Einzelheiten befreit, habe ich versucht, das Datenformat einfach genug zu gestalten, damit sie es verstehen können, falls ich morgen plötzlich verschwinde…

Die Geschwindigkeit ist deutlich gestiegen. Normalerweise scannen wir funktional signifikante Fragmente des Genoms. Früher konnten wir das nicht tun (es war zu teuer), aber jetzt, dank der Gruppenstruktur (Bin) und dem Caching, benötigt eine Anfrage von einem SNP im Durchschnitt weniger als 0,1 Sekunden, und der Datenverbrauch ist so gering, dass die Kosten für S3 minimal sind.

Kürzlich wurde ich mit der Verwaltung von über 25 TB Rohdaten zur Genotypisierung in meinem Labor beauftragt. Als ich anfing, benötigte ich 8 Minuten und es kostete 20 Dollar, um einen SNP abzufragen. Nach der Verwendung von AWK + #rstats verarbeitet es jetzt weniger als den Zehntel einer Sekunde und kostet 0,00001 Dollar. Mein persönlicher #BigData Erfolg. pic.twitter.com/ANOXVGrmkk

11. März 2019 30. Mai 2019

Fazit

Dieser Artikel ist keineswegs ein Leitfaden. Die Lösung war individuell und vermutlich nicht optimal. Vielmehr ist es eine Erzählung über eine Reise. Ich möchte, dass andere verstehen, dass solche Lösungen nicht vollständig ausgebildet im Kopf entstehen, sondern das Ergebnis von Versuch und Irrtum sind. Wenn Sie nach einem Datenanalysten suchen, bedenken Sie, dass man Erfahrung benötigt, um diese Werkzeuge effektiv zu nutzen, und Erfahrung kostet Geld. Ich bin froh, dass ich Mittel zur Verfügung hatte, um es zu bezahlen, aber viele andere, die die gleiche Arbeit besser machen könnten als ich, haben nie die Möglichkeit, dies wegen fehlender Mittel auch nur zu versuchen.

Die Werkzeuge für Big Data sind vielseitig. Wenn Sie Zeit haben, können Sie wahrscheinlich eine schnellere Lösung schreiben, indem Sie ‚intelligente‘ Datenbereinigung, Speicherung und Auswahlmethoden anwenden. Letztendlich geht es um die Analyse von Kosten und Nutzen.

Was ich gelernt habe:

• Es gibt keinen günstigen Weg, 25 TB auf einmal zu parsen;
• Seien Sie vorsichtig mit der Größe Ihrer Parquet-Dateien und deren Organisation;
• Partitionen in Spark müssen ausgewogen sein;
• Versuchen Sie niemals, 2,5 Millionen Partitionen zu erstellen;
• Sortieren bleibt schwierig, ebenso wie das Einrichten von Spark;
• Manchmal erfordern spezielle Daten spezielle Lösungen;
• Das Zusammenführen in Spark funktioniert schnell, aber das Partitionieren bleibt teuer;
• Schlafen Sie nicht, während Ihnen die Grundlagen beigebracht werden, sicherlich hat schon jemand Ihr Problem in den 1980er Jahren gelöst;
• gnu parallel Es ist eine magische Sache, jeder sollte sie nutzen;
• Spark liebt unkomprimierte Daten und mag es nicht, Partitionen zu kombinieren;
• In Spark gibt es zu viel Overhead bei der Lösung einfacher Aufgaben;
• Assoziative Arrays in AWK sind sehr effizient;
• Man kann auf stdin und stdout aus R-Skripten zugreifen und es somit in der Pipeline verwenden;
• Dank intelligenter Pfadimplementierung kann S3 viele Dateien verarbeiten;
• Der Hauptgrund für Zeitverschwendung ist die vorzeitige Optimierung Ihrer Speicherstrategie;
• Versuchen Sie nicht, Aufgaben manuell zu optimieren, lassen Sie das der Computer machen;
• Die API sollte einfach sein, um Einfachheit und Flexibilität in der Nutzung zu gewährleisten;
• Wenn Ihre Daten gut aufbereitet sind, wird das cachen einfach sein!

Quelle: habr.com