Πώς να διαβάσετε αυτό το άρθρο: Ζητώ συγγνώμη που το κείμενο ήταν τόσο μεγάλο και χαοτικό. Για να εξοικονομήσετε χρόνο, ξεκινώ κάθε κεφάλαιο με μια εισαγωγή «Τι έμαθα», η οποία συνοψίζει την ουσία του κεφαλαίου σε μία ή δύο προτάσεις.

«Απλώς δείξε μου τη λύση!» Εάν θέλετε απλώς να δείτε από πού ήρθα, μεταβείτε στο κεφάλαιο «Γίνομαι πιο εφευρετικός», αλλά νομίζω ότι είναι πιο ενδιαφέρον και χρήσιμο να διαβάζεις για την αποτυχία.

Μου ανατέθηκε πρόσφατα να δημιουργήσω μια διαδικασία για την επεξεργασία ενός μεγάλου όγκου ακατέργαστων αλληλουχιών DNA (τεχνικά ένα τσιπ SNP). Η ανάγκη ήταν να ληφθούν γρήγορα δεδομένα σχετικά με μια δεδομένη γενετική τοποθεσία (που ονομάζεται SNP) για επακόλουθη μοντελοποίηση και άλλες εργασίες. Χρησιμοποιώντας τα R και AWK, κατάφερα να καθαρίσω και να οργανώσω τα δεδομένα με φυσικό τρόπο, επιταχύνοντας σημαντικά την επεξεργασία των ερωτημάτων. Αυτό δεν ήταν εύκολο για μένα και απαιτούσε πολλές επαναλήψεις. Αυτό το άρθρο θα σας βοηθήσει να αποφύγετε μερικά από τα λάθη μου και θα σας δείξει σε τι κατέληξα.

Πρώτα, μερικές εισαγωγικές εξηγήσεις.

Δεδομένα

Το πανεπιστημιακό μας κέντρο επεξεργασίας γενετικών πληροφοριών μας παρείχε δεδομένα με τη μορφή TSV 25 TB. Τα έλαβα χωρισμένα σε 5 πακέτα συμπιεσμένα με Gzip, το καθένα από τα οποία περιείχε περίπου 240 αρχεία τεσσάρων gigabyte. Κάθε σειρά περιείχε δεδομένα για ένα SNP από ένα άτομο. Συνολικά, διαβιβάστηκαν στοιχεία για ~2,5 εκατομμύρια SNP και ~60 χιλιάδες άτομα. Εκτός από τις πληροφορίες SNP, τα αρχεία περιείχαν πολλές στήλες με αριθμούς που αντικατοπτρίζουν διάφορα χαρακτηριστικά, όπως ένταση ανάγνωσης, συχνότητα διαφορετικών αλληλόμορφων κ.λπ. Συνολικά υπήρχαν περίπου 30 στήλες με μοναδικές τιμές.

στόχος

Όπως με κάθε έργο διαχείρισης δεδομένων, το πιο σημαντικό πράγμα ήταν να καθοριστεί πώς θα χρησιμοποιηθούν τα δεδομένα. Στην προκειμένη περίπτωση θα επιλέξουμε κυρίως μοντέλα και ροές εργασίας για SNP με βάση το SNP. Δηλαδή, θα χρειαζόμαστε δεδομένα μόνο για ένα SNP κάθε φορά. Έπρεπε να μάθω πώς να ανακτώ όλες τις εγγραφές που σχετίζονται με ένα από τα 2,5 εκατομμύρια SNP όσο το δυνατόν πιο εύκολα, γρήγορα και φθηνότερα.

Πώς να μην το κάνετε αυτό

Για να παραθέσω ένα κατάλληλο κλισέ:

Δεν απέτυχα χίλιες φορές, απλώς ανακάλυψα χίλιους τρόπους για να αποφύγω την ανάλυση μιας δέσμης δεδομένων σε μορφή φιλική προς τα ερωτήματα.

Πρώτη προσπάθεια

Τι έχω μάθει: Δεν υπάρχει φθηνός τρόπος ανάλυσης 25 TB κάθε φορά.

Έχοντας παρακολουθήσει το μάθημα «Προηγμένες Μέθοδοι για Επεξεργασία Μεγάλων Δεδομένων» στο Πανεπιστήμιο Vanderbilt, ήμουν σίγουρος ότι το κόλπο ήταν στο σακίδιο. Θα χρειαστούν πιθανώς μία ή δύο ώρες για να ρυθμίσετε τον διακομιστή Hive ώστε να τρέχει μέσω όλων των δεδομένων και να αναφέρει το αποτέλεσμα. Δεδομένου ότι τα δεδομένα μας είναι αποθηκευμένα στο AWS S3, χρησιμοποίησα την υπηρεσία Αθήνα, το οποίο σας επιτρέπει να εφαρμόζετε ερωτήματα Hive SQL σε δεδομένα S3. Δεν χρειάζεται να δημιουργήσετε/δημιουργήσετε ένα σύμπλεγμα Hive και επίσης πληρώνετε μόνο για τα δεδομένα που αναζητάτε.

Αφού έδειξα στην Athena τα δεδομένα μου και τη μορφή τους, έκανα μερικές δοκιμές με ερωτήματα όπως αυτό:

select * from intensityData limit 10;

Και έλαβε γρήγορα καλά δομημένα αποτελέσματα. Ετοιμος.

Μέχρι που προσπαθήσαμε να χρησιμοποιήσουμε τα δεδομένα στη δουλειά μας...

Μου ζητήθηκε να βγάλω όλες τις πληροφορίες SNP για να δοκιμάσω το μοντέλο. Έτρεξα το ερώτημα:

select * from intensityData 
where snp = 'rs123456';

...και άρχισε να περιμένει. Μετά από οκτώ λεπτά και περισσότερα από 4 TB ζητούμενων δεδομένων, έλαβα το αποτέλεσμα. Η Athena χρεώνει με βάση τον όγκο των δεδομένων που βρέθηκαν, 5 $ ανά terabyte. Έτσι, αυτό το μεμονωμένο αίτημα κόστιζε 20 $ και οκτώ λεπτά αναμονής. Για να τρέξουμε το μοντέλο σε όλα τα δεδομένα, έπρεπε να περιμένουμε 38 χρόνια και να πληρώσουμε 50 εκατομμύρια δολάρια Προφανώς, αυτό δεν ήταν κατάλληλο για εμάς.

Ήταν απαραίτητο να χρησιμοποιήσετε παρκέ...

Τι έχω μάθει: Να είστε προσεκτικοί με το μέγεθος των αρχείων Parquet και την οργάνωσή τους.

Πρώτα προσπάθησα να διορθώσω την κατάσταση μετατρέποντας όλα τα TSV σε Αρχεία παρκέ. Είναι βολικά για εργασία με μεγάλα σύνολα δεδομένων επειδή οι πληροφορίες σε αυτά αποθηκεύονται σε στήλη: κάθε στήλη βρίσκεται στη δική της μνήμη/τμήμα δίσκου, σε αντίθεση με τα αρχεία κειμένου, στα οποία οι σειρές περιέχουν στοιχεία κάθε στήλης. Και αν πρέπει να βρείτε κάτι, τότε απλώς διαβάστε την απαιτούμενη στήλη. Επιπλέον, κάθε αρχείο αποθηκεύει ένα εύρος τιμών σε μια στήλη, οπότε αν η τιμή που αναζητάτε δεν βρίσκεται στο εύρος της στήλης, το Spark δεν θα χάσει χρόνο σαρώνοντας ολόκληρο το αρχείο.

Έκανα μια απλή εργασία Κόλλα AWS για να μετατρέψουμε τα TSV μας σε Parquet και ρίξαμε τα νέα αρχεία στο Athena. Χρειάστηκαν περίπου 5 ώρες. Αλλά όταν έτρεξα το αίτημα, χρειάστηκε περίπου ο ίδιος χρόνος και λίγο λιγότερα χρήματα για να ολοκληρωθεί. Το γεγονός είναι ότι το Spark, προσπαθώντας να βελτιστοποιήσει την εργασία, απλά αποσυσκευάστηκε ένα κομμάτι TSV και το έβαλε στο δικό του κομμάτι Parquet. Και επειδή κάθε κομμάτι ήταν αρκετά μεγάλο ώστε να περιέχει ολόκληρες τις εγγραφές πολλών ανθρώπων, κάθε αρχείο περιείχε όλα τα SNP, οπότε το Spark έπρεπε να ανοίξει όλα τα αρχεία για να εξαγάγει τις πληροφορίες που χρειαζόταν.

Είναι ενδιαφέρον ότι ο προεπιλεγμένος (και συνιστώμενος) τύπος συμπίεσης του Parquet, snappy, δεν μπορεί να διαχωριστεί. Επομένως, κάθε εκτελεστής είχε κολλήσει στην εργασία της αποσυσκευασίας και λήψης του πλήρους συνόλου δεδομένων 3,5 GB.

Ας καταλάβουμε το πρόβλημα

Τι έχω μάθει: Η ταξινόμηση είναι δύσκολη, ειδικά εάν τα δεδομένα διανέμονται.

Μου φάνηκε ότι τώρα κατάλαβα την ουσία του προβλήματος. Χρειάστηκε μόνο να ταξινομήσω τα δεδομένα κατά στήλη SNP, όχι κατά άτομα. Στη συνέχεια, πολλά SNP θα αποθηκευτούν σε ξεχωριστό κομμάτι δεδομένων και, στη συνέχεια, η "έξυπνη" λειτουργία του Parquet "ανοίγει μόνο εάν η τιμή είναι εντός του εύρους" θα εμφανιστεί σε όλο της το μεγαλείο. Δυστυχώς, η ταξινόμηση δισεκατομμυρίων σειρών διάσπαρτων σε ένα σύμπλεγμα αποδείχθηκε δύσκολη εργασία.

Παρακολουθώ το μάθημα αλγορίθμων στο κολέγιο: "Ουφ, κανείς δεν ενδιαφέρεται για την υπολογιστική πολυπλοκότητα όλων αυτών των αλγορίθμων ταξινόμησης"

Προσπαθώ να κάνω ταξινόμηση σε στήλη 20 TB #σπίθα πίνακας: "Γιατί διαρκεί τόσο πολύ;" #Επιστημονικά δεδομένα αγώνες.

— Nick Strayer (@NicholasStrayer) 11 Μαρτίου 2019

Η AWS σίγουρα δεν θέλει να εκδώσει επιστροφή χρημάτων λόγω του λόγου "Είμαι αποσπασμένος φοιτητής". Αφού έτρεξα με ταξινόμηση στο Amazon Glue, έτρεξε για 2 ημέρες και έπεσε.

Τι γίνεται με την κατάτμηση;

Τι έχω μάθει: Τα χωρίσματα στο Spark πρέπει να είναι ισορροπημένα.

Στη συνέχεια σκέφτηκα την ιδέα της κατάτμησης δεδομένων σε χρωμοσώματα. Υπάρχουν 23 από αυτά (και πολλά άλλα αν λάβετε υπόψη το μιτοχονδριακό DNA και τις μη χαρτογραφημένες περιοχές).
Αυτό θα σας επιτρέψει να χωρίσετε τα δεδομένα σε μικρότερα κομμάτια. Εάν προσθέσετε μόνο μία γραμμή στη συνάρτηση εξαγωγής Spark στη δέσμη ενεργειών Glue partition_by = "chr", τότε τα δεδομένα πρέπει να χωριστούν σε κουβάδες.

Το γονιδίωμα αποτελείται από πολλά θραύσματα που ονομάζονται χρωμοσώματα.

Δυστυχώς, δεν λειτούργησε. Τα χρωμοσώματα έχουν διαφορετικά μεγέθη, που σημαίνει διαφορετικές ποσότητες πληροφοριών. Αυτό σημαίνει ότι οι εργασίες που έστειλε το Spark στους εργαζόμενους δεν ήταν ισορροπημένες και ολοκληρώθηκαν αργά, επειδή ορισμένοι κόμβοι τελείωσαν νωρίς και ήταν αδρανείς. Ωστόσο, οι εργασίες ολοκληρώθηκαν. Αλλά όταν ζητήθηκε ένα SNP, η ανισορροπία προκάλεσε και πάλι προβλήματα. Το κόστος επεξεργασίας των SNP σε μεγαλύτερα χρωμοσώματα (δηλαδή εκεί που θέλουμε να πάρουμε δεδομένα) έχει μειωθεί μόνο κατά περίπου έναν παράγοντα 10. Πολλά, αλλά όχι αρκετά.

Κι αν το χωρίσουμε σε ακόμα μικρότερα μέρη;

Τι έχω μάθει: Μην προσπαθήσετε ποτέ να κάνετε 2,5 εκατομμύρια κατατμήσεις καθόλου.

Αποφάσισα να πάω όλα έξω και χώρισα σε κάθε SNP. Αυτό εξασφάλισε ότι τα χωρίσματα ήταν ίσου μεγέθους. ΗΤΑΝ ΜΙΑ ΚΑΚΗ ΙΔΕΑ. Χρησιμοποίησα κόλλα και πρόσθεσα μια αθώα γραμμή partition_by = 'snp'. Η εργασία ξεκίνησε και άρχισε να εκτελείται. Μια μέρα αργότερα έλεγξα και είδα ότι δεν υπήρχε ακόμα τίποτα γραμμένο στο S3, οπότε σκότωσα την εργασία. Φαίνεται ότι ο Glue έγραφε ενδιάμεσα αρχεία σε μια κρυφή τοποθεσία στο S3, πολλά αρχεία, ίσως μερικά εκατομμύρια. Ως αποτέλεσμα, το λάθος μου κόστισε περισσότερα από χίλια δολάρια και δεν άρεσε στον μέντορά μου.

Διαμέριση + ταξινόμηση

Τι έχω μάθει: Η ταξινόμηση είναι ακόμα δύσκολη, όπως και ο συντονισμός του Spark.

Η τελευταία μου απόπειρα διαμερίσματος περιελάμβανε την κατάτμηση των χρωμοσωμάτων και στη συνέχεια την ταξινόμηση κάθε διαμερίσματος. Θεωρητικά, αυτό θα επιτάχυνε κάθε ερώτημα επειδή τα επιθυμητά δεδομένα SNP έπρεπε να βρίσκονται μέσα σε μερικά κομμάτια Parquet μέσα σε ένα δεδομένο εύρος. Δυστυχώς, η ταξινόμηση ακόμη και διαμερισμένων δεδομένων αποδείχθηκε δύσκολη υπόθεση. Ως αποτέλεσμα, άλλαξα σε EMR για ένα προσαρμοσμένο σύμπλεγμα και χρησιμοποίησα οκτώ ισχυρές παρουσίες (C5.4xl) και Sparklyr για να δημιουργήσω μια πιο ευέλικτη ροή εργασίας...

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

...ωστόσο, το έργο δεν ολοκληρώθηκε ακόμη. Το ρύθμισα με διαφορετικούς τρόπους: αύξησα την κατανομή μνήμης για κάθε εκτελεστή ερωτήματος, χρησιμοποίησα κόμβους με μεγάλη ποσότητα μνήμης, χρησιμοποίησα μεταβλητές εκπομπής (μεταβλητές εκπομπής), αλλά κάθε φορά αυτές αποδεικνύονταν ημίμετρα και σταδιακά άρχισαν οι εκτελεστές να αποτύχει μέχρι να σταματήσουν όλα.

Ενημέρωση: έτσι ξεκινά. pic.twitter.com/agY4GU2ru5

— Nick Strayer (@NicholasStrayer) 15 Μαΐου 2019

Γίνομαι πιο δημιουργικός

Τι έχω μάθει: Μερικές φορές τα ειδικά δεδομένα απαιτούν ειδικές λύσεις.

Κάθε SNP έχει μια τιμή θέσης. Αυτός είναι ένας αριθμός που αντιστοιχεί στον αριθμό των βάσεων κατά μήκος του χρωμοσώματος του. Αυτός είναι ένας ωραίος και φυσικός τρόπος να οργανώνουμε τα δεδομένα μας. Στην αρχή ήθελα να χωρίσω ανά περιοχές κάθε χρωμοσώματος. Για παράδειγμα, θέσεις 1 - 2000, 2001 - 4000 κ.λπ. Αλλά το πρόβλημα είναι ότι τα SNP δεν είναι ομοιόμορφα κατανεμημένα στα χρωμοσώματα, επομένως τα μεγέθη των ομάδων θα ποικίλλουν πολύ.

Ως αποτέλεσμα, κατέληξα σε μια κατανομή των θέσεων σε κατηγορίες (κατάταξη). Χρησιμοποιώντας τα ήδη ληφθέντα δεδομένα, υπέβαλα ένα αίτημα για να αποκτήσω μια λίστα με μοναδικά SNP, τις θέσεις και τα χρωμοσώματά τους. Στη συνέχεια ταξινόμησα τα δεδομένα σε κάθε χρωμόσωμα και συνέλεξα SNPs σε ομάδες (bin) συγκεκριμένου μεγέθους. Ας πούμε 1000 SNP το καθένα. Αυτό μου έδωσε τη σχέση SNP-προς-ομάδα-ανά-χρωμόσωμα.

Στο τέλος έφτιαξα ομάδες (bin) των 75 SNP, ο λόγος θα εξηγηθεί παρακάτω.

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

Πρώτα δοκιμάστε με το Spark

Τι έχω μάθει: Η συγκέντρωση σπινθήρων είναι γρήγορη, αλλά η κατάτμηση εξακολουθεί να είναι ακριβή.

Ήθελα να διαβάσω αυτό το μικρό πλαίσιο δεδομένων (2,5 εκατομμυρίων σειρών) στο Spark, να το συνδυάσω με τα μη επεξεργασμένα δεδομένα και, στη συνέχεια, να το χωρίσω κατά τη στήλη που προστέθηκε πρόσφατα 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')
  )

χρησιμοποίησα sdf_broadcast(), οπότε το Spark γνωρίζει ότι θα πρέπει να στείλει το πλαίσιο δεδομένων σε όλους τους κόμβους. Αυτό είναι χρήσιμο εάν τα δεδομένα είναι μικρού μεγέθους και απαιτούνται για όλες τις εργασίες. Διαφορετικά, το Spark προσπαθεί να είναι έξυπνο και διανέμει δεδομένα όπως χρειάζεται, κάτι που μπορεί να προκαλέσει επιβράδυνση.

Και πάλι, η ιδέα μου δεν λειτούργησε: τα καθήκοντα λειτούργησαν για κάποιο χρονικό διάστημα, ολοκλήρωσαν την ένωση και στη συνέχεια, όπως οι εκτελεστές που ξεκίνησαν από τη διαίρεση, άρχισαν να αποτυγχάνουν.

Προσθήκη AWK

Τι έχω μάθει: Μην κοιμάστε όταν σας διδάσκουν τα βασικά. Σίγουρα κάποιος έχει ήδη λύσει το πρόβλημά σας στη δεκαετία του 1980.

Μέχρι αυτό το σημείο, ο λόγος για όλες τις αποτυχίες μου με το Spark ήταν το συνονθύλευμα δεδομένων στο σύμπλεγμα. Ίσως η κατάσταση μπορεί να βελτιωθεί με την προκαταρκτική θεραπεία. Αποφάσισα να προσπαθήσω να χωρίσω τα δεδομένα ακατέργαστου κειμένου σε στήλες χρωμοσωμάτων, οπότε ήλπιζα να παρέχω στο Spark δεδομένα "προκατατμημένα".

Έψαξα στο StackOverflow για τον τρόπο διαχωρισμού ανά τιμές στήλης και βρήκα μια τόσο μεγάλη απάντηση. Με το AWK μπορείτε να χωρίσετε ένα αρχείο κειμένου ανά τιμές στήλης γράφοντάς το σε ένα σενάριο αντί να στείλετε τα αποτελέσματα σε stdout.

Έγραψα ένα σενάριο Bash για να το δοκιμάσω. Κατέβασε ένα από τα συσκευασμένα TSV και, στη συνέχεια, το αποσυσκευάστηκε χρησιμοποιώντας gzip και εστάλη σε awk.

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

Δούλεψε!

Γέμισμα των πυρήνων

Τι έχω μάθει: gnu parallel - Είναι ένα μαγικό πράγμα, όλοι πρέπει να το χρησιμοποιούν.

Ο χωρισμός ήταν αρκετά αργός και όταν ξεκίνησα htopγια να ελέγξω τη χρήση μιας ισχυρής (και ακριβής) παρουσίας EC2, αποδείχθηκε ότι χρησιμοποιούσα μόνο έναν πυρήνα και περίπου 200 MB μνήμης. Για να λύσουμε το πρόβλημα και να μην χάσουμε πολλά χρήματα, έπρεπε να καταλάβουμε πώς να παραλληλίσουμε το έργο. Ευτυχώς, σε ένα απολύτως εκπληκτικό βιβλίο Επιστήμη δεδομένων στη γραμμή εντολών Βρήκα ένα κεφάλαιο του Jeron Janssens για την παραλληλοποίηση. Από αυτό έμαθα gnu parallel, μια πολύ ευέλικτη μέθοδος για την υλοποίηση multithreading στο Unix.

Όταν ξεκίνησα τη διαμέριση χρησιμοποιώντας τη νέα διαδικασία, όλα ήταν εντάξει, αλλά υπήρχε ακόμα ένα σημείο συμφόρησης - η λήψη αντικειμένων S3 στο δίσκο δεν ήταν πολύ γρήγορη και δεν ήταν πλήρως παραλληλισμένη. Για να το διορθώσω, έκανα αυτό:

1. Ανακάλυψα ότι είναι δυνατή η εφαρμογή του σταδίου λήψης S3 απευθείας στη διοχέτευση, εξαλείφοντας εντελώς την ενδιάμεση αποθήκευση στο δίσκο. Αυτό σημαίνει ότι μπορώ να αποφύγω την εγγραφή ακατέργαστων δεδομένων στο δίσκο και να χρησιμοποιήσω ακόμη μικρότερο, και επομένως φθηνότερο, χώρο αποθήκευσης στο AWS.
2. Ομάδα aws configure set default.s3.max_concurrent_requests 50 αύξησε σημαντικά τον αριθμό των νημάτων που χρησιμοποιεί το AWS CLI (από προεπιλογή υπάρχουν 10).
3. Πέρασα σε μια παρουσία EC2 βελτιστοποιημένη για ταχύτητα δικτύου, με το γράμμα n στο όνομα. Διαπίστωσα ότι η απώλεια επεξεργαστικής ισχύος κατά τη χρήση n-στιγμών αντισταθμίζεται περισσότερο από την αύξηση της ταχύτητας φόρτωσης. Για τις περισσότερες εργασίες χρησιμοποίησα c5n.4xl.
4. Άλλαξε gzip επί pigz, αυτό είναι ένα εργαλείο gzip που μπορεί να κάνει ωραία πράγματα για να παραλληλίσει την αρχικά μη παραλληλισμένη εργασία της αποσυμπίεσης αρχείων (αυτό βοήθησε λιγότερο).

# 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

Αυτά τα βήματα συνδυάζονται μεταξύ τους για να λειτουργήσουν όλα πολύ γρήγορα. Αυξάνοντας τις ταχύτητες λήψης και εξαλείφοντας τις εγγραφές στο δίσκο, θα μπορούσα τώρα να επεξεργαστώ ένα πακέτο 5 terabyte σε λίγες μόνο ώρες.

Δεν υπάρχει τίποτα πιο γλυκό από το να βλέπεις να χρησιμοποιούνται όλοι οι πυρήνες για τους οποίους πληρώνεις στο AWS. Χάρη στο gnu-parallel μπορώ να αποσυμπιέσω και να χωρίσω ένα csv 19 gig τόσο γρήγορα όσο μπορώ να το κατεβάσω. Δεν μπορούσα να βρω ούτε σπινθήρα για να το τρέξω αυτό. #Επιστημονικά δεδομένα #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) 17 Μαΐου 2019

Αυτό το tweet θα έπρεπε να αναφέρει το "TSV". Αλίμονο.

Χρησιμοποιώντας πρόσφατα αναλυμένα δεδομένα

Τι έχω μάθει: Στον Spark αρέσουν τα ασυμπίεστα δεδομένα και δεν του αρέσει ο συνδυασμός κατατμήσεων.

Τώρα τα δεδομένα ήταν στο S3 σε μη συσκευασμένη (ανάγνωση: κοινόχρηστη) και ημι-παραγγελμένη μορφή και μπορούσα να επιστρέψω ξανά στο Spark. Με περίμενε μια έκπληξη: Και πάλι δεν κατάφερα να πετύχω αυτό που ήθελα! Ήταν πολύ δύσκολο να πούμε στο Spark πώς ακριβώς έγινε η κατάτμηση των δεδομένων. Και ακόμη και όταν το έκανα αυτό, αποδείχθηκε ότι υπήρχαν πάρα πολλά διαμερίσματα (95 χιλιάδες) και όταν χρησιμοποίησα coalesce μείωσε τον αριθμό τους σε λογικά όρια, αυτό κατέστρεψε το διαμέρισμα μου. Είμαι σίγουρος ότι αυτό μπορεί να διορθωθεί, αλλά μετά από μερικές μέρες αναζήτησης δεν μπόρεσα να βρω λύση. Τελικά τελείωσα όλες τις εργασίες στο Spark, αν και χρειάστηκε λίγος χρόνος και τα διαχωρισμένα αρχεία Parquet δεν ήταν πολύ μικρά (~200 KB). Ωστόσο, τα δεδομένα ήταν εκεί που χρειάζονταν.

Πολύ μικρό και ανώμαλο, υπέροχο!

Δοκιμή τοπικών ερωτημάτων Spark

Τι έχω μάθει: Το Spark έχει υπερβολικά μεγάλο κόστος κατά την επίλυση απλών προβλημάτων.

Κάνοντας λήψη των δεδομένων σε μια έξυπνη μορφή, μπόρεσα να δοκιμάσω την ταχύτητα. Ρυθμίστε μια δέσμη ενεργειών R για την εκτέλεση ενός τοπικού διακομιστή Spark και, στη συνέχεια, φορτώστε ένα πλαίσιο δεδομένων Spark από τον καθορισμένο χώρο αποθήκευσης ομάδας Parquet (κάδο). Προσπάθησα να φορτώσω όλα τα δεδομένα αλλά δεν κατάφερα να αναγνωρίσω το Sparklyr το διαμέρισμα.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

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

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

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

print(Sys.time() - start_time)

Η εκτέλεση διήρκεσε 29,415 δευτερόλεπτα. Πολύ καλύτερο, αλλά όχι πολύ καλό για μαζικές δοκιμές οτιδήποτε. Επιπλέον, δεν μπορούσα να επιταχύνω τα πράγματα με την προσωρινή αποθήκευση, επειδή όταν προσπαθούσα να αποθηκεύσω προσωρινά ένα πλαίσιο δεδομένων στη μνήμη, το Spark κατέρρεε πάντα, ακόμη και όταν εκχωρούσα περισσότερα από 50 GB μνήμης σε ένα σύνολο δεδομένων που ζύγιζε λιγότερο από 15.

Επιστροφή στο AWK

Τι έχω μάθει: Οι συσχετιστικοί πίνακες στο AWK είναι πολύ αποδοτικοί.

Συνειδητοποίησα ότι μπορούσα να πετύχω μεγαλύτερες ταχύτητες. Το θυμήθηκα σε ένα υπέροχο Εκμάθηση AWK από τον Bruce Barnett Διάβασα για ένα ωραίο χαρακτηριστικό που ονομάζεται "συσχετιστικούς πίνακες" Ουσιαστικά, αυτά είναι ζεύγη κλειδιού-τιμής, τα οποία για κάποιο λόγο ονομάζονταν διαφορετικά στο AWK και επομένως κατά κάποιο τρόπο δεν τα σκέφτηκα πολύ. Ρομάν Τσεπλιάκα υπενθύμισε ότι ο όρος «συσχετιστικοί πίνακες» είναι πολύ παλαιότερος από τον όρο «ζεύγος κλειδιού-τιμής». Ακόμα κι αν εσύ αναζητήστε το κλειδί-τιμή στο Google Ngram, δεν θα δείτε αυτόν τον όρο εκεί, αλλά θα βρείτε συσχετιστικούς πίνακες! Επιπλέον, το "ζεύγος κλειδιού-τιμής" συνδέεται συχνότερα με βάσεις δεδομένων, επομένως είναι πολύ πιο λογικό να το συγκρίνουμε με ένα hashmap. Συνειδητοποίησα ότι μπορούσα να χρησιμοποιήσω αυτούς τους συσχετιστικούς πίνακες για να συσχετίσω τα SNP μου με έναν πίνακα bin και ακατέργαστα δεδομένα χωρίς να χρησιμοποιήσω το Spark.

Για να γίνει αυτό, στο σενάριο AWK χρησιμοποίησα το μπλοκ BEGIN. Αυτό είναι ένα κομμάτι κώδικα που εκτελείται πριν περάσει η πρώτη γραμμή δεδομένων στο κύριο σώμα του σεναρίου.

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

Ομάδα while(getline...) φόρτωσαν όλες τις σειρές από την ομάδα CSV (bin), ορίστε την πρώτη στήλη (όνομα SNP) ως κλειδί για τον συσχετιστικό πίνακα bin και η δεύτερη τιμή (ομάδα) ως τιμή. Μετά στο μπλοκ { }, το οποίο εκτελείται σε όλες τις γραμμές του κύριου αρχείου, κάθε γραμμή αποστέλλεται στο αρχείο εξόδου, το οποίο λαμβάνει ένα μοναδικό όνομα ανάλογα με την ομάδα του (bin): ..._bin_"bin[$1]"_....

Μεταβλητές batch_num и chunk_id ταίριαξε με τα δεδομένα που παρέχονται από το pipeline, αποφεύγοντας μια συνθήκη αγώνα και κάθε νήμα εκτέλεσης που εκτελείται parallel, έγραψε στο δικό του μοναδικό αρχείο.

Δεδομένου ότι διασκόρπισα όλα τα ακατέργαστα δεδομένα σε φακέλους σε χρωμοσώματα που είχαν απομείνει από το προηγούμενο πείραμά μου με το AWK, τώρα θα μπορούσα να γράψω ένα άλλο σενάριο Bash για να επεξεργάζομαι ένα χρωμόσωμα τη φορά και να στέλνω βαθύτερα διαμερισμένα δεδομένα στο S3.

DESIRED_CHR='13'

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

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

Το σενάριο έχει δύο ενότητες parallel.

Στην πρώτη ενότητα, διαβάζονται δεδομένα από όλα τα αρχεία που περιέχουν πληροφορίες για το επιθυμητό χρωμόσωμα, στη συνέχεια αυτά τα δεδομένα διανέμονται σε νήματα, τα οποία διανέμουν τα αρχεία στις κατάλληλες ομάδες (bin). Για να αποφευχθούν οι συνθήκες κούρσας όταν πολλά νήματα γράφουν στο ίδιο αρχείο, το AWK μεταβιβάζει τα ονόματα των αρχείων για να γράψει δεδομένα σε διαφορετικά μέρη, π.χ. chr_10_bin_52_batch_2_aa.csv. Ως αποτέλεσμα, δημιουργούνται πολλά μικρά αρχεία στο δίσκο (για αυτό χρησιμοποίησα τόμους EBS terabyte).

Μεταφορέας από το δεύτερο τμήμα parallel περνά μέσα από τις ομάδες (bin) και συνδυάζει τα μεμονωμένα αρχεία τους σε κοινό CSV c catκαι μετά τα στέλνει για εξαγωγή.

Εκπομπή στο R;

Τι έχω μάθει: Μπορείτε να επικοινωνήσετε stdin и stdout από μια δέσμη ενεργειών R, και ως εκ τούτου χρησιμοποιήστε την στην ολοκλήρωση.

Μπορεί να έχετε παρατηρήσει αυτή τη γραμμή στο σενάριο Bash: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Μεταφράζει όλα τα συνδεδεμένα ομαδικά αρχεία (bin) στο σενάριο R παρακάτω. {} είναι μια ειδική τεχνική parallel, το οποίο εισάγει τα δεδομένα που στέλνει στην καθορισμένη ροή απευθείας στην ίδια την εντολή. Επιλογή {#} παρέχει ένα μοναδικό αναγνωριστικό νήματος και {%} αντιπροσωπεύει τον αριθμό της θέσης εργασίας (επαναλαμβανόμενο, αλλά ποτέ ταυτόχρονα). Μπορείτε να βρείτε μια λίστα με όλες τις επιλογές τεκμηρίωση.

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

Όταν μια μεταβλητή file("stdin") μεταδίδεται σε readr::read_csv, τα δεδομένα που μεταφράζονται στο σενάριο R φορτώνονται σε ένα πλαίσιο, το οποίο στη συνέχεια βρίσκεται στη μορφή .rds-αρχείο με χρήση aws.s3 γραμμένο απευθείας στο S3.

Το RDS είναι κάτι σαν μια junior έκδοση του Parquet, χωρίς τις λεπτομέρειες της αποθήκευσης ηχείων.

Αφού τελείωσα το σενάριο του Bash, έλαβα ένα πακέτο .rds-αρχεία που βρίσκονται στο S3, τα οποία μου επέτρεψαν να χρησιμοποιήσω αποτελεσματική συμπίεση και ενσωματωμένους τύπους.

Παρά τη χρήση του φρένου R, όλα λειτούργησαν πολύ γρήγορα. Δεν αποτελεί έκπληξη το γεγονός ότι τα μέρη του R που διαβάζουν και γράφουν δεδομένα είναι εξαιρετικά βελτιστοποιημένα. Μετά τη δοκιμή σε ένα μεσαίου μεγέθους χρωμόσωμα, η εργασία ολοκληρώθηκε σε μια παρουσία C5n.4xl σε περίπου δύο ώρες.

Περιορισμοί S3

Τι έχω μάθει: Χάρη στην εφαρμογή έξυπνης διαδρομής, το S3 μπορεί να χειριστεί πολλά αρχεία.

Ανησυχούσα αν το S3 θα μπορούσε να χειριστεί τα πολλά αρχεία που μεταφέρθηκαν σε αυτό. Θα μπορούσα να κάνω τα ονόματα των αρχείων να έχουν νόημα, αλλά πώς θα τα αναζητούσε το S3;

Οι φάκελοι στο S3 είναι απλώς για επίδειξη, στην πραγματικότητα το σύστημα δεν ενδιαφέρεται για το σύμβολο /. Από τη σελίδα Συχνές ερωτήσεις του S3.

Φαίνεται ότι το S3 αντιπροσωπεύει τη διαδρομή προς ένα συγκεκριμένο αρχείο ως ένα απλό κλειδί σε ένα είδος κατακερματισμού ή βάσης δεδομένων που βασίζεται σε έγγραφα. Ένας κάδος μπορεί να θεωρηθεί ως πίνακας και τα αρχεία μπορούν να θεωρηθούν εγγραφές σε αυτόν τον πίνακα.

Δεδομένου ότι η ταχύτητα και η αποτελεσματικότητα είναι σημαντικές για την απόκτηση κέρδους στο Amazon, δεν αποτελεί έκπληξη το γεγονός ότι αυτό το σύστημα διαδρομής κλειδιού ως αρχείου έχει βελτιστοποιηθεί φρικτά. Προσπάθησα να βρω μια ισορροπία: έτσι ώστε να μην χρειάζεται να κάνω πολλά αιτήματα λήψης, αλλά τα αιτήματα να εκτελούνται γρήγορα. Αποδείχθηκε ότι είναι καλύτερο να δημιουργήσετε περίπου 20 χιλιάδες αρχεία κάδου. Νομίζω ότι αν συνεχίσουμε να βελτιστοποιούμε, μπορούμε να επιτύχουμε αύξηση της ταχύτητας (για παράδειγμα, να φτιάξουμε έναν ειδικό κάδο μόνο για δεδομένα, μειώνοντας έτσι το μέγεθος του πίνακα αναζήτησης). Αλλά δεν υπήρχε χρόνος ή χρήμα για περαιτέρω πειράματα.

Τι γίνεται με τη διασταυρούμενη συμβατότητα;

Τι έμαθα: Η υπ' αριθμόν ένα αιτία χαμένου χρόνου είναι η πρόωρη βελτιστοποίηση της μεθόδου αποθήκευσης.

Σε αυτό το σημείο, είναι πολύ σημαντικό να αναρωτηθείτε: "Γιατί να χρησιμοποιήσετε μια ιδιόκτητη μορφή αρχείου;" Ο λόγος έγκειται στην ταχύτητα φόρτωσης (τα αρχεία CSV με gzip χρειάστηκε 7 φορές περισσότερο χρόνο για να φορτωθούν) και στη συμβατότητα με τις ροές εργασίας μας. Μπορεί να ξανασκεφτώ εάν το R μπορεί εύκολα να φορτώσει αρχεία Parquet (ή Arrow) χωρίς το Spark load. Όλοι στο εργαστήριό μας χρησιμοποιούν το R, και αν χρειαστεί να μετατρέψω τα δεδομένα σε άλλη μορφή, εξακολουθώ να έχω τα αρχικά δεδομένα κειμένου, οπότε μπορώ απλώς να εκτελέσω ξανά τη διοχέτευση.

Καταμερισμός εργασιών

Τι έχω μάθει: Μην προσπαθείτε να βελτιστοποιήσετε τις εργασίες χειροκίνητα, αφήστε τον υπολογιστή να το κάνει.

Έχω διορθώσει τη ροή εργασίας σε ένα χρωμόσωμα, τώρα πρέπει να επεξεργαστώ όλα τα άλλα δεδομένα.
Ήθελα να ανεβάσω αρκετές περιπτώσεις EC2 για μετατροπή, αλλά ταυτόχρονα φοβόμουν μήπως πάρω ένα πολύ μη ισορροπημένο φορτίο σε διαφορετικές εργασίες επεξεργασίας (όπως ακριβώς το Spark υπέφερε από μη ισορροπημένα διαμερίσματα). Επιπλέον, δεν με ενδιέφερε να ανεβάσω μία παρουσία ανά χρωμόσωμα, επειδή για λογαριασμούς AWS υπάρχει προεπιλεγμένο όριο 10 περιπτώσεων.

Στη συνέχεια αποφάσισα να γράψω ένα σενάριο σε R για να βελτιστοποιήσω τις εργασίες επεξεργασίας.

Πρώτα, ζήτησα από το S3 να υπολογίσει πόσο χώρο αποθήκευσης καταλάμβανε κάθε χρωμόσωμα.

library(aws.s3)
library(tidyverse)

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



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

Μετά έγραψα μια συνάρτηση που παίρνει το συνολικό μέγεθος, ανακατεύει τη σειρά των χρωμοσωμάτων, τα χωρίζει σε ομάδες num_jobs και σας λέει πόσο διαφορετικά είναι τα μεγέθη όλων των εργασιών επεξεργασίας.

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

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

shuffle_job(1)



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

Έπειτα έτρεξα μέσα από χίλιες ανακατώσεις χρησιμοποιώντας purrr και διάλεξα το καλύτερο.

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

Έτσι κατέληξα με ένα σύνολο εργασιών που ήταν πολύ παρόμοια σε μέγεθος. Στη συνέχεια, το μόνο που έμεινε ήταν να τυλίξω το προηγούμενο σενάριο μου στο Bash σε ένα μεγάλο βρόχο for. Αυτή η βελτιστοποίηση χρειάστηκε περίπου 10 λεπτά για να γραφτεί. Και αυτό είναι πολύ λιγότερο από ό,τι θα ξόδευα για τη μη αυτόματη δημιουργία εργασιών, εάν δεν ήταν ισορροπημένες. Επομένως, νομίζω ότι είχα δίκιο με αυτήν την προκαταρκτική βελτιστοποίηση.

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

Στο τέλος προσθέτω την εντολή shutdown:

sudo shutdown -h now

... και όλα πήγαν καλά! Χρησιμοποιώντας το AWS CLI, ανέφερα περιπτώσεις χρησιμοποιώντας την επιλογή user_data τους έδωσε σενάρια Bash των εργασιών τους για επεξεργασία. Έτρεχαν και έκλεισαν αυτόματα, επομένως δεν πλήρωνα για επιπλέον ισχύ επεξεργασίας.

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

Ας μαζέψουμε!

Τι έχω μάθει: Το API θα πρέπει να είναι απλό για λόγους ευκολίας και ευελιξίας στη χρήση.

Τελικά πήρα τα δεδομένα στη σωστή θέση και μορφή. Το μόνο που απέμενε ήταν να απλοποιήσω όσο το δυνατόν περισσότερο τη διαδικασία χρήσης δεδομένων για να διευκολύνω τους συναδέλφους μου. Ήθελα να φτιάξω ένα απλό API για τη δημιουργία αιτημάτων. Αν στο μέλλον αποφασίσω να αλλάξω από .rds σε αρχεία Parquet, τότε αυτό θα πρέπει να είναι πρόβλημα για μένα, όχι για τους συναδέλφους μου. Για αυτό αποφάσισα να φτιάξω ένα εσωτερικό πακέτο R.

Δημιουργήστε και τεκμηριώστε ένα πολύ απλό πακέτο που περιέχει μερικές μόνο λειτουργίες πρόσβασης δεδομένων οργανωμένες γύρω από μια λειτουργία get_snp. Έκανα επίσης μια ιστοσελίδα για τους συναδέλφους μου pkgdown, ώστε να μπορούν να δουν εύκολα παραδείγματα και τεκμηρίωση.

Έξυπνη προσωρινή αποθήκευση

Τι έχω μάθει: Εάν τα δεδομένα σας είναι καλά προετοιμασμένα, η προσωρινή αποθήκευση θα είναι εύκολη!

Εφόσον μία από τις κύριες ροές εργασίας εφάρμοσε το ίδιο μοντέλο ανάλυσης στο πακέτο SNP, αποφάσισα να χρησιμοποιήσω το binning προς όφελός μου. Κατά τη μετάδοση δεδομένων μέσω SNP, όλες οι πληροφορίες από την ομάδα (bin) επισυνάπτονται στο επιστρεφόμενο αντικείμενο. Δηλαδή, τα παλιά ερωτήματα μπορούν (θεωρητικά) να επιταχύνουν την επεξεργασία νέων ερωτημάτων.

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

Κατά τη δημιουργία του πακέτου, έτρεξα πολλά σημεία αναφοράς για να συγκρίνω την ταχύτητα όταν χρησιμοποιούσα διαφορετικές μεθόδους. Συνιστώ να μην το αμελήσετε αυτό, γιατί μερικές φορές τα αποτελέσματα είναι απροσδόκητα. Για παράδειγμα, dplyr::filter ήταν πολύ πιο γρήγορη από τη λήψη σειρών με χρήση φιλτραρίσματος βάσει ευρετηρίου και η ανάκτηση μιας στήλης από ένα φιλτραρισμένο πλαίσιο δεδομένων ήταν πολύ πιο γρήγορη από τη χρήση της σύνταξης ευρετηρίασης.

Σημειώστε ότι το αντικείμενο prev_snp_results περιέχει το κλειδί snps_in_bin. Αυτή είναι μια σειρά από όλα τα μοναδικά SNP σε μια ομάδα (bin), που σας επιτρέπει να ελέγξετε γρήγορα εάν έχετε ήδη δεδομένα από ένα προηγούμενο ερώτημα. Επίσης, διευκολύνει τον βρόχο σε όλα τα SNP σε μια ομάδα (bin) με αυτόν τον κωδικό:

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

Ευρήματα

Τώρα μπορούμε (και έχουμε αρχίσει να τρέχουμε σοβαρά) μοντέλα και σενάρια που προηγουμένως ήταν απρόσιτα για εμάς. Το καλύτερο είναι ότι οι συνάδελφοί μου στο εργαστήριο δεν χρειάζεται να σκεφτούν τυχόν επιπλοκές. Απλώς έχουν μια λειτουργία που λειτουργεί.

Και παρόλο που το πακέτο δεν τους δίνει τις λεπτομέρειες, προσπάθησα να κάνω τη μορφή δεδομένων αρκετά απλή ώστε να μπορούν να το καταλάβουν αν εξαφανιστώ ξαφνικά αύριο...

Η ταχύτητα έχει αυξηθεί αισθητά. Συνήθως σαρώνουμε λειτουργικά σημαντικά θραύσματα γονιδιώματος. Προηγουμένως, δεν μπορούσαμε να το κάνουμε αυτό (αποδείχτηκε ότι ήταν πολύ ακριβό), αλλά τώρα, χάρη στη δομή της ομάδας (bin) και την προσωρινή αποθήκευση, ένα αίτημα για ένα SNP διαρκεί κατά μέσο όρο λιγότερο από 0,1 δευτερόλεπτα και η χρήση δεδομένων είναι τόσο χαμηλό ότι το κόστος για το S3 είναι φιστίκια.

Πρόσφατα μπήκα σε αντιπαράθεση 25+ TB ακατέργαστων δεδομένων γονότυπου για το εργαστήριό μου. Όταν ξεκίνησα, η χρήση του spark χρειάστηκε 8 λεπτά και κόστισε 20 $ για να ρωτήσω ένα SNP. Μετά τη χρήση του AWK+ #rstats για την επεξεργασία, χρειάζεται πλέον λιγότερο από ένα 10ο του δευτερολέπτου και κοστίζει 0.00001 $. Το προσωπικό μου #Μεγάλα δεδομένα νίκη. pic.twitter.com/ANOXVGrmkk

— Nick Strayer (@NicholasStrayer) 30 Μαΐου 2019

Συμπέρασμα

Αυτό το άρθρο δεν είναι καθόλου οδηγός. Η λύση αποδείχθηκε ατομική και σχεδόν σίγουρα όχι βέλτιστη. Μάλλον είναι ένα ταξιδιωτικό. Θέλω να καταλάβουν οι άλλοι ότι τέτοιες αποφάσεις δεν φαίνονται πλήρως διαμορφωμένες στο κεφάλι, είναι αποτέλεσμα δοκιμής και λάθους. Επίσης, αν ψάχνετε για έναν επιστήμονα δεδομένων, να έχετε κατά νου ότι η αποτελεσματική χρήση αυτών των εργαλείων απαιτεί εμπειρία και η εμπειρία κοστίζει χρήματα. Είμαι χαρούμενος που είχα τα μέσα να πληρώσω, αλλά πολλοί άλλοι που μπορούν να κάνουν την ίδια δουλειά καλύτερα από εμένα δεν θα έχουν ποτέ την ευκαιρία λόγω έλλειψης χρημάτων να προσπαθήσουν καν.

Τα εργαλεία μεγάλων δεδομένων είναι ευέλικτα. Εάν έχετε χρόνο, μπορείτε σχεδόν σίγουρα να γράψετε μια πιο γρήγορη λύση χρησιμοποιώντας έξυπνες τεχνικές καθαρισμού, αποθήκευσης και εξαγωγής δεδομένων. Τελικά καταλήγει σε μια ανάλυση κόστους-οφέλους.

Τι έμαθα:

• δεν υπάρχει φθηνός τρόπος ανάλυσης 25 TB κάθε φορά.
• Να είστε προσεκτικοί με το μέγεθος των αρχείων Parquet και την οργάνωσή τους.
• Τα χωρίσματα στο Spark πρέπει να είναι ισορροπημένα.
• Γενικά, μην προσπαθήσετε ποτέ να κάνετε 2,5 εκατομμύρια κατατμήσεις.
• Η ταξινόμηση είναι ακόμα δύσκολη, όπως και η ρύθμιση του Spark.
• Μερικές φορές τα ειδικά δεδομένα απαιτούν ειδικές λύσεις.
• Η συγκέντρωση σπινθήρων είναι γρήγορη, αλλά η κατάτμηση εξακολουθεί να είναι ακριβή.
• Μην κοιμάστε όταν σας διδάσκουν τα βασικά, κάποιος πιθανότατα έχει ήδη λύσει το πρόβλημά σας στη δεκαετία του 1980.
• gnu parallel - αυτό είναι ένα μαγικό πράγμα, όλοι πρέπει να το χρησιμοποιήσουν.
• Στον Spark αρέσουν τα ασυμπίεστα δεδομένα και δεν του αρέσει ο συνδυασμός κατατμήσεων.
• Το Spark έχει πάρα πολλά έξοδα κατά την επίλυση απλών προβλημάτων.
• Οι συσχετιστικοί πίνακες του AWK είναι πολύ αποδοτικοί.
• μπορείτε να επικοινωνήσετε stdin и stdout από μια δέσμη ενεργειών R, και ως εκ τούτου χρησιμοποιήστε την στη διαδικασία.
• Χάρη στην εφαρμογή έξυπνης διαδρομής, το S3 μπορεί να επεξεργαστεί πολλά αρχεία.
• Ο κύριος λόγος για τη σπατάλη χρόνου είναι η πρόωρη βελτιστοποίηση της μεθόδου αποθήκευσης.
• Μην προσπαθήσετε να βελτιστοποιήσετε τις εργασίες χειροκίνητα, αφήστε τον υπολογιστή να το κάνει.
• Το API θα πρέπει να είναι απλό για λόγους ευκολίας και ευελιξίας στη χρήση.
• Εάν τα δεδομένα σας είναι καλά προετοιμασμένα, η προσωρινή αποθήκευση θα είναι εύκολη!

Πηγή: www.habr.com