Jak czytać ten artykuł: przepraszam, że tekst wyszedł tak długi i chaotyczny. Żeby zaoszczędzić twój czas, każdą rozdział zaczynam od wprowadzenia "Czego się nauczyłem", w którym w jednym lub dwóch zdaniach przedstawiam sedno rozdziału.

"Po prostu pokaż rozwiązanie!" Jeśli chcesz tylko zobaczyć, do czego doszedłem, przejdź do rozdziału "Staję się bardziej pomysłowy", ale uważam, że ciekawiej i użyteczniej jest poczytać o niepowodzeniach.

Niedawno zlecono mi skonfigurowanie procesu przetwarzania dużej ilości sekwencji DNA (technicznie to chip SNP). Musiałem szybko uzyskać dane o zadanym miejscu genetycznym (które nazywa się SNP) do dalszego modelowania i innych zadań. Dzięki R i AWK udało mi się naturalnie oczyścić i zorganizować dane, znacznie przyspieszając przetwarzanie zapytań. Nie było to łatwe i wymagało wielu iteracji. Ten artykuł pomoże ci uniknąć niektórych moich błędów i pokaże, co ostatecznie udało mi się osiągnąć.

Na początek kilka wprowadzeń.

Dane

Nasz uniwersytecki centrum przetwarzania informacji genetycznej dostarczyło nam dane w formacie TSV o objętości 25 TB. Otrzymałem je rozbite na 5 paczek, skompresowanych Gzip, z których każda zawierała około 240 czterogigabajtowych plików. Każdy wiersz zawierał dane dla jednego SNP jednej osoby. Łącznie przesłano dane dla ~2,5 miliona SNP i ~60 tysięcy osób. Oprócz informacji SNP w plikach znajdowało się wiele kolumn z liczbami, odzwierciedlającymi różne cechy, takie jak intensywność odczytu, częstotliwość różnych alleli itd. Było około 30 kolumn z unikalnymi wartościami.

Cel

Jak w każdym projekcie przetwarzania danych, najważniejsze było określenie, jak dane będą wykorzystywane. W tym przypadku w znacznej mierze będziemy dobierać modele i procesy robocze dla SNP na podstawie SNP. To znaczy, jednocześnie potrzebne będą dane tylko dla jednego SNP. Musiałem nauczyć się jak najprościej, najszybciej i najtaniej wydobywać wszystkie rekordy związane z jednym z 2,5 miliona SNP.

Jak tego nie robić

Zacytuję odpowiednie klisze:

Nie poniosłem porażki tysiąc razy, po prostu odkryłem tysiąc sposobów, aby nie parsować dużej ilości danych w formacie użytecznym do zapytań.

Pierwsza próba

Czego się nauczyłem: nie ma taniego sposobu, aby przetworzyć 25 TB za jednym razem.

Po wysłuchaniu wykładu na Uniwersytecie Vanderbilta zatytułowanego „Zaawansowane metody przetwarzania dużych danych”, byłem pewien, że to prosta sprawa. Może zajmie to godzinkę lub dwie, aby skonfigurować serwer Hive, by przeszukać wszystkie dane i złożyć raport. Athena, który pozwala na stosowanie zapytań Hive SQL do danych w S3. Nie trzeba konfigurować/uruchamiać klastra Hive, a płacisz tylko za te dane, których szukasz.

Po tym, jak pokazałem Athena moje dane i ich format, uruchomiłem kilka testów z podobnymi zapytaniami:

select * from intensityData limit 10;

I szybko otrzymałem dobrze zorganizowane wyniki. Gotowe.

Dopóki nie spróbowaliśmy użyć danych w pracy…

Poproszono mnie o wyciągnięcie wszelkich informacji o SNP, aby przetestować na nich model. Uruchomiłem zapytanie:

select * from intensityData 
where snp = 'rs123456';

…i zacząłem czekać. Po ośmiu minutach i ponad 4 TB żądanych danych otrzymałem rezultat. Athena pobiera opłatę za ilość znalezionych danych, 5 dolarów za terabajt. Tak więc to jedno zapytanie kosztowało 20 dolarów i osiem minut oczekiwania. Aby uruchomić model na wszystkich danych, trzeba by czekać 38 lat i zapłacić 50 milionów dolarów. Oczywiście, to nam nie odpowiadało.

Należało użyć Parquet…

Czego się nauczyłem: bądź ostrożny z rozmiarem swoich plików Parquet i ich organizacją.

Na początku próbowałem poprawić sytuację, konwertując wszystkie pliki TSV na pliki Parquet.Są one wygodne do pracy z dużymi zbiorami danych, ponieważ informacje w nich są przechowywane w formie kolumnowej: każda kolumna leży w własnym segmencie pamięci/dysku, w przeciwieństwie do plików tekstowych, w których wiersze zawierają elementy każdej kolumny. Jeśli trzeba coś znaleźć, wystarczy przeczytać potrzebną kolumnę. Dodatkowo w każdym pliku w kolumnie przechowywany jest zakres wartości, więc jeśli poszukiwany element nie mieści się w zakresie kolumny, Spark nie będzie tracił czasu na skanowanie całego pliku.

Uruchomiłem prostą zadanie AWS Glue Aby przekonwertować nasze pliki TSV na Parquet, przesłałem nowe pliki do Athena. Zajęło to około 5 godzin. Jednak gdy uruchomiłem zapytanie, jego wykonanie zajęło mniej więcej tyle samo czasu i trochę mniej pieniędzy. Problem w tym, że Spark, próbując zoptymalizować zadanie, po prostu rozpakował jeden chunk TSV i umieścił go w swoim własnym chunku Parquet. A ponieważ każdy chunk był dość duży i zawierał pełne rekordy wielu osób, w każdym pliku przechowywane były wszystkie SNP, więc Spark musiał otworzyć wszystkie pliki, aby wydobyć potrzebne informacje.

Praw ciekawostką jest, że domyślny (i zalecany) typ kompresji w Parquet – snappy – nie jest dzielony (splitable). Dlatego każdy wykonawca (executor) utknął na zadaniu rozpakowywania i ładowania pełnego zestawu danych o wielkości 3,5 GB.

Zgłębiamy problem

Czego się nauczyłem: sortowanie jest trudne, zwłaszcza gdy dane są rozproszone.

Wydawało mi się, że teraz rozumiem istotę problemu. Musiałem tylko posortować dane według kolumny SNP, a nie według osób. Wtedy w osobnym chunku danych będzie przechowywanych kilka SNP, a wtedy w całej okazałości ujawni się „inteligentna” funkcja Parquet „otwieraj tylko, jeśli wartość leży w zakresie”. Niestety, posortowanie miliardów wierszy porozrzucanych po klastrze okazało się trudnym zadaniem.

Ja na zajęciach z algorytmów w college'u: «Ech, nikt się nie przejmuje złożonością obliczeniową tych wszystkich algorytmów sortowania»

Ja próbujący sortować kolumnę w 20TB #spark tabeli: «Dlaczego to trwa tak długo?» #DataScience trudności.

— Nick Strayer (@NicholasStrayer) 11 marca 2019

AWS z pewnością nie chce zwrócić pieniędzy z powodu „Jestem roztrzepanym studentem”. Po uruchomieniu sortowania na Amazon Glue, trwało to 2 dni i zakończyło się niepowodzeniem.

Jak to wygląda z partycjonowaniem?

Czego się nauczyłem: partycje w Sparku muszą być zbalansowane.

Potem przyszło mi do głowy, aby partycjonować dane według chromosomów. Jest ich 23 (i kilka więcej, jeśli uwzględnimy mitochondrialne DNA i nieprzypisane (unmapped) obszary).
To pozwoli podzielić dane na mniejsze porcje. Jeśli dodam do funkcji eksportu w Spark w skrypcie Glue tylko jeden wiersz partition_by = "chr", dane powinny być rozłożone na kubły (buckets).

Genom składa się z licznych fragmentów, które nazywane są chromosomami.

Niestety, to nie zadziałało. Chromosomy mają różne rozmiary, co oznacza, że zawierają różną ilość informacji. To sprawia, że zadania, które Spark wysyłał do pracowników, nie były zbalansowane i wykonywały się wolno, ponieważ niektóre węzły kończyły wcześniej i pozostawały bezczynne. Jednak zadania zostały wykonane. Lecz przy zapytaniu o jeden SNP nierównowaga znów stała się przyczyną problemów. Koszt przetwarzania SNP w większych chromosomach (czyli tam, skąd chcemy uzyskać dane) zmniejszył się jedynie około dziesięciokrotnie. Dużo, ale nie wystarczająco.

A co jeśli podzielić na jeszcze mniejsze partycje?

Czego się nauczyłem: nigdy w życiu nie próbuj tworzyć 2,5 miliona partycji.

Postanowiłem pójść na całość i podzielić każdy SNP. To zapewniło jednolity rozmiar partycji. TO BYŁ ZŁY POMYSŁ. Skorzystałem z Glue i dodałem niewinną linię partition_by = 'snp'. Zadanie uruchomiło się i zaczęło działać. Dzień później sprawdziłem i zobaczyłem, że w S3 nadal nic nie zapisano, więc zabiłem zadanie. Wygląda na to, że Glue zapisywał pliki pośrednie w ukryte miejsce w S3, i było ich dużo, możliwe że kilka milionów. W rezultacie mój błąd kosztował ponad tysiąc dolarów i nie ucieszył mojego mentora.

Partycjonowanie + sortowanie

Czego się nauczyłem: wciąż trudno jest sortować, podobnie jak konfigurować Sparka.

Ostatnia próba partycjonowania polegała na tym, że partycjonowałem chromosomy, a następnie sortowałem każdą partycję. W teorii miałoby to przyspieszyć każde zapytanie, ponieważ oczekiwane dane o SNP powinny znajdować się w kilku blokach Parquet w określonym zakresie. Niestety, sortowanie nawet partycjonowanych danych okazało się trudnym zadaniem. W rezultacie przeszedłem na EMR dla własnego klastra i użyłem ośmiu mocnych instancji (C5.4xl) oraz Sparklyr do stworzenia bardziej elastycznego przepływu pracy…

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

…jednak zadanie nadal nie zostało wykonane. Konfigurowałem na wiele sposobów: zwiększałem przydział pamięci dla każdego wykonawcy zapytań, używałem węzłów z większą ilością pamięci, stosowałem zmienne rozsyłane (broadcasting variables), ale za każdym razem były to półśrodki i z czasem wykonawcy zaczęli się psuć, aż wszystko się zatrzymało.

Aktualizacja: wszystko się zaczyna. pic.twitter.com/agY4GU2ru5

— Nick Strayer (@NicholasStrayer) 15 maja 2019

Staję się bardziej pomysłowy

Czego się nauczyłem: czasami szczególne dane wymagają szczególnych rozwiązań.

Każdy SNP ma wartość pozycji. To liczba odpowiadająca liczbie podstaw, które znajdują się wzdłuż jego chromosomu. To dobry i naturalny sposób organizacji naszych danych. Na początku chciałem podzielić dane według obszarów każdego chromosomu. Na przykład, pozycje 1 — 2000, 2001 — 4000 itd. Ale problem w tym, że SNP są rozmieszczone na chromosomach nierównomiernie, dlatego rozmiar grup będzie bardzo różny.

W rezultacie doszedłem do podziału według kategorii (rank) pozycji. Na już załadowanych danych uruchomiłem zapytanie, aby uzyskać listę unikalnych SNP, ich pozycji i chromosomów. Następnie posortowałem dane w każdym chromosomie i zgrupowałem SNP w grupy (bin) o określonym rozmiarze. Powiedzmy, po 1000 SNP. To dało mi powiązanie SNP z grupą w chromosomie.

Ostatecznie stworzyłem grupy (bin) po 75 SNP, co wyjaśnię poniżej.

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

Pierwsza próba z Spark

Czego się nauczyłem: łączenie w Spark działa szybko, ale partycjonowanie wciąż jest kosztowne.

Chciałem wczytać ten mały (2,5 miliona wierszy) zbiór danych do Sparka, połączyć go z surowymi danymi, a następnie partycjonować według nowo dodanej kolumny. 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')
  )

Użyłem sdf_broadcast(), w ten sposób Spark wie, że musi przesłać zbiór danych do wszystkich węzłów. Jest to przydatne, jeśli dane są małego rozmiaru i są wymagane dla wszystkich zadań. W przeciwnym razie Spark stara się być sprytny i rozdziela dane w miarę potrzeby, co może prowadzić do opóźnień.

I znów mój pomysł nie zadziałał: zadania przez jakiś czas działały, kończyły łączenie, a następnie, podobnie jak uruchomieni wykonawcy partycjonowania, zaczęły kończyć się błędami.

Dodaję AWK

Czego się nauczyłem: nie zasypiaj podczas nauki podstaw. Na pewno ktoś rozwiązał twój problem już w latach 80.

Do tego momentu przyczyną wszystkich moich niepowodzeń z Spark był nieład danych w klastrze. Możliwe, że sytuację można poprawić dzięki wstępnej obróbce. Postanowiłem spróbować podzielić surowe dane tekstowe na kolumny chromosomów, mając nadzieję, że dostarczę Sparkowi 'wstępnie partycjonowane' dane.

Szukalem na StackOverflow, jak dzielić według wartości kolumn, i znalazłem taki piękny odpowiedź. Dzięki AWK możesz podzielić plik tekstowy według wartości kolumn, wykonując zapis w skrypcie, a nie przesyłając wyniki do stdout.

Dla przykładu napisałem skrypt Bash. Ściągnąłem jeden z zapakowanych plików TSV, a następnie rozpakowałem go za pomocą gzip i wysłałem do awk.

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

To zadziałało!

Wypełnianie rdzeni

Czego się nauczyłem: gnu parallel to magiczna rzecz, wszyscy powinni jej używać.

Podział przebiegał dość wolno, a kiedy uruchomiłem htop, aby sprawdzić użycie potężnego (i drogiego) instancji EC2, okazało się, że wykorzystuję tylko jeden rdzeń i około 200 MB pamięci. Aby rozwiązać problem i nie stracić dużo pieniędzy, musiałem wymyślić, jak zrównoleglić pracę. Na szczęście w niesamowitej książce Data Science at the Command Line Jeroena Janssensa znalazłem rozdział poświęcony równoległości. Dowiedziałem się o gnu parallel, bardzo elastycznej metodzie realizacji wielowątkowości w Unixie.

Gdy uruchomiłem podział za pomocą nowego procesu, wszystko działało świetnie, ale pozostawało wąskie gardło — pobieranie obiektów S3 na dysk nie było zbyt szybkie i nie było w pełni zrównoleglone. Aby to naprawić, zrobiłem następujące:

1. Odkryłem, że można bezpośrednio w potoku zrealizować etap pobierania S3, całkowicie rezygnując z pośredniego przechowywania na dysku. Oznacza to, że mogę uniknąć zapisywania surowych danych na dysku i wykorzystać jeszcze mniejsze, a więc i tańsze miejsce na AWS.
2. Komendą aws configure set default.s3.max_concurrent_requests 50 znacząco zwiększyło liczbę wątków, które wykorzystuje AWS CLI (domyślnie jest ich 10).
3. Przeszedłem na zoptymalizowaną pod kątem prędkości sieci instancję EC2 z literą n w nazwie. Odkryłem, że strata mocy obliczeniowej przy użyciu instancji n jest w pełni kompensowana przez wzrost prędkości pobierania. Do większości zadań używałem c5n.4xl.
4. Zmieniłem gzip na pigz, to narzędzie gzip, które umie robić fajne rzeczy w celu zrównoleglenia pierwotnie nierozdzielnego zadania rozpakowywania plików (to pomogło najmniej).

# 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

Te kroki zostały ze sobą połączone, aby wszystko działało bardzo szybko. Dzięki zwiększonej prędkości pobierania i rezygnacji z zapisywania na dysku mogłem teraz przetworzyć 5-terabajtowy pakiet w zaledwie kilka godzin.

Nie ma nic słodszego niż widok wszystkich rdzeni, za które płacisz na AWS, wykorzystywanych w pełni. Dzięki gnu-parallel mogę rozpakować i podzielić 19-gigowy plik CSV tak szybko, jak go pobieram. Nawet nie mogłem uruchomić Sparka. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) 17 maja 2019

W tym tweecie miało być wspomniane o 'TSV'. Niestety.

Ponowne wykorzystanie przetworzonych danych

Czego się nauczyłem: Spark lubi dane w formacie niesk compressed i nie lubi łączyć partycji.

Teraz dane znajdowały się w S3 w nieskompresowanym (czytaj, rozdzielanym) i półposortowanym formacie, a ja mogłem wrócić do Sparka. Czekała mnie niespodzianka: znów nie udało mi się osiągnąć oczekiwanych wyników! Było bardzo trudno dokładnie powiedzieć Sparkowi, jak są partycjonowane dane. I nawet gdy to zrobiłem, okazało się, że partycji jest za dużo (95 tys.), a gdy przy pomocy coalesce zmniejszyłem ich liczbę do rozsądnych granic, zaburzyło to moje partycjonowanie. Jestem pewien, że można to naprawić, ale przez kilka dni poszukiwań nie udało mi się znaleźć rozwiązania. Ostatecznie udało mi się zrealizować wszystkie zadania w Sparku, chociaż zajęło to trochę czasu, a moje podzielone pliki Parquet nie były za małe (~200 KB). Niemniej jednak dane znajdowały się tam, gdzie powinny.

Zbyt małe i niejednakowe, świetnie!

Testowanie lokalnych zapytań w Sparku

Czego się nauczyłem: W Sparku jest zbyt wiele narzutów przy rozwiązywaniu prostych zadań.

Wczytując dane w przemyślanym formacie, mogłem przetestować szybkość. Ustawiłem skrypt w R do uruchomienia lokalnego serwera Sparka, a następnie załadowałem zbiór danych Sparka z określonego magazynu grup Parquet (bin). Próbowałem wczytać wszystkie dane, ale nie mogłem sprawić, by Sparklyr rozpoznał partycjonowanie.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Rozpocznij timer
start_time <- Sys.time()

# Wczytaj pożądany bin do Sparka
intensity_data % 
  Spark_read_Parquet(
    name = 'intensity_data', 
    path = get_snp_location(desired_snp),
    memory = FALSE )

# Podziel bin na snp, a następnie zbierz lokalnie
test_subset % 
  filter(SNP_Name == desired_snp) %>% 
  collect()

print(Sys.time() - start_time)

Wykonanie zajęło 29,415 sekundy. Znacznie lepiej, ale nadal niezbyt dobrze do masowego testowania czegokolwiek. Co więcej, nie mogłem przyspieszyć działania dzięki pamięci podręcznej, ponieważ gdy próbowałem przechować ramkę danych w pamięci, Spark zawsze się zawieszał, nawet gdy przydzieliłem ponad 50 GB pamięci dla zbioru danych, który ważył mniej niż 15.

Powrót do AWK

Czego się nauczyłem: tablice asocjacyjne w AWK są bardzo wydajne.

Zdałem sobie sprawę, że mogę osiągnąć większą prędkość. Przypomniałem sobie o znakomitym przewodniku po AWK Bruce'a Barnetta Czytałem o świetnej funkcji, która nazywa się „tablice asocjacyjne”. W zasadzie to pary klucz-wartość, które w AWK zostały z jakiegoś powodu nazwane inaczej, dlatego jakoś o nich szczególnie nie przypominałem. Roman Cheplyaka przypomniał, że termin „tablice asocjacyjne” jest znacznie starszy niż termin „para klucz-wartość”. Nawet jeśli poszukasz klucz-wartość w Google Ngram, tego terminu tam nie znajdziesz, ale znajdziesz tablice asocjacyjne! Ponadto „para klucz-wartość” najczęściej kojarzy się z bazami danych, więc znacznie logiczniej porównywać je z hashmap. Zrozumiałem, że mogę używać tych tablic asocjacyjnych do powiązania moich SNP z tabelą grup (bin table) oraz surowymi danymi bez użycia Sparka.

W tym celu w skrypcie AWK użyłem bloku BEGIN. To fragment kodu, który jest wykonywany, zanim pierwszy wiersz danych zostanie przekazany do głównej części skryptu.

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

Zespół while(getline...) załadował wszystkie wiersze z grupy CSV (bin), ustawił pierwszą kolumnę (nazwa SNP) jako klucz dla tablicy asocjacyjnej bin i drugą wartość (grupa) jako wartość. Następnie w bloku { }, który jest stosowany do wszystkich wierszy głównego pliku, każdy wiersz jest wysyłany do pliku wyjściowego, który otrzymuje unikalną nazwę w zależności od swojej grupy (bin): ..._bin_"bin[$1]"_....

Zmienne batch_num i chunk_id były zgodne z danymi dostarczonymi przez potok, co pozwoliło uniknąć stanu wyścigu, a każdy wątek wykonawczy uruchomiony równolegle, pisał w własnym unikalnym pliku.

Ponieważ wszystkie surowe dane rozrzuciłem po folderach według chromosomów, które pozostały po moim poprzednim eksperymencie z AWK, mogłem teraz napisać inny skrypt Bash, aby przetwarzać po jednym chromosomie i przesyłać do S3 głębiej partycjonowane dane.

DESIRED_CHR='13'

# Pobierz dane chromosomu z s3 i podziel je na biny
aws s3 ls $DATA_LOC |
awk '{print $4}' |
grep 'chr'$DESIRED_CHR'.csv' |
parallel "echo 'czytanie {}'; aws s3 cp "$DATA_LOC"{} - | awk -v chr=""$DESIRED_CHR"" -v chunk="{}" -f split_on_chr_bin.awk"

# Połącz wszystkie równoległe fragmenty do pojedynczych plików i załaduj do rds używając R
ls chunked/ |
cut -d '_' -f 4 |
sort -u |
parallel "echo 'pakowanie bin {}'; cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R '$S3_DEST'/chr_'$DESIRED_CHR'_bin_{}.rds"
rm chunked/*

W skrypcie są dwa sekcje równolegle.

W pierwszej sekcji dane są odczytywane ze wszystkich plików zawierających informacje o danej chromosomie, a następnie te dane są rozdzielane na wątki, które przypisują pliki do odpowiednich grup (bin). Aby uniknąć stanu wyścigu, w którym kilka wątków zapisuje do jednego pliku, AWK przekazuje nazwy plików do zapisywania danych w różnych miejscach, na przykład, chr_10_bin_52_batch_2_aa.csv. W rezultacie na dysku powstaje wiele małych plików (do tego użyłem terabajtowych wolumenów EBS).

Drugi sekcja równolegle przechodzi przez grupy (bin) i łączy ich pojedyncze pliki w ogólne CSV z cat, a następnie przesyła je do eksportu.

Translacja w R?

Czego się nauczyłem: można do niej uzyskać dostęp z stdin i stdout skryptu R, co oznacza, że można go również wykorzystać w łańcuchu przetwarzania.

W skrypcie Bash mogłeś zauważyć taki wiersz: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Przekazuje on wszystkie konkatenowane pliki grupy (bin) do poniższego skryptu R. {} jest specjalną metodą równolegle, która wszelkie dane przesyłane do wskazanego wątku wstawia bezpośrednio do samej komendy. Opcja {#} zapewnia unikalny identyfikator wątku wykonania, a {%} to numer slotu zadania (powtarza się, ale nigdy jednocześnie). Listę wszystkich opcji można znaleźć w dokumentacji.

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

Kiedy zmienna file("stdin") jest przekazywana do readr::read_csv, dane przekazane do skryptu R są ładowane do ramki, która potem w postaci .rds-pliku za pomocą aws.s3 jest zapisywana bezpośrednio do S3.

RDS to coś w rodzaju młodszej wersji Parquet, bez wyrafinowania magazynu kolumnowego.

Po zakończeniu skryptu Bash otrzymałem paczkę .rds-plików znajdujących się w S3, co pozwoliło mi na efektywne kompresowanie i wbudowane typy.

Pomimo użycia wolnego R, wszystko działało bardzo szybko. Nie ma się czemu dziwić, fragmenty w R odpowiedzialne za odczyt i zapis danych są dobrze zoptymalizowane. Po testowaniu na jednej chromosomie średniej wielkości, zadanie zostało ukończone na instancji C5n.4xl w około dwie godziny.

Ograniczenia S3

Czego się nauczyłem: dzięki inteligentnej implementacji ścieżek S3 może obsługiwać wiele plików.

Obawiałem się, czy S3 poradzi sobie z dużą liczbą przesyłanych jej plików. Mogłem nadać im sensowne nazwy, ale jak S3 będzie po nich wyszukiwać?

Foldery w S3 są tylko dla estetyki, system w rzeczywistości nie interesuje symbol /. Z FAQ S3.

Wygląda na to, że S3 przedstawia ścieżkę do konkretnego pliku w postaci prostego klucza w swego rodzaju tabeli haszującej lub bazie danych opartej na dokumentach. Koszyk (bucket) można traktować jak tabelę, a pliki jak rekordy w tej tabeli.

Ponieważ prędkość i wydajność mają kluczowe znaczenie dla zysków w Amazonie, nie jest zaskoczeniem, że ten system 'klucz-w-ścieżce-do-pliku' jest niesamowicie zoptymalizowany. Starałem się znaleźć równowagę: aby nie było potrzeby wykonywania wielu zapytań get, ale jednocześnie, aby zapytania były szybkie. Okazało się, że najlepiej składać około 20 tys. plików binarnych. Myślę, że jeśli będę kontynuował optymalizację, można osiągnąć wzrost prędkości (na przykład tworząc specjalny koszyk tylko dla danych, co zmniejszy rozmiar tabeli wyszukiwania). Jednak na dalsze eksperymenty nie było już czasu ani pieniędzy.

A co z interoperacyjnością?

Czego się nauczyłem: głównym powodem marnotrawienia czasu jest przedwczesna optymalizacja twojej metody przechowywania.

W tym momencie ważne jest, aby zadać sobie pytanie: 'Dlaczego używać własnościowego formatu plików?' Powodem jest szybkość ładowania (skompresowane pliki CSV gzip ładowały się 7 razy dłużej) oraz zgodność z naszymi procesami roboczymi. Mogę ponownie rozważyć swoje decyzje, jeśli R będzie w stanie łatwo ładować pliki Parquet (lub Arrow) bez obciążenia w postaci Spark. W naszym laboratorium wszyscy używają R, a jeśli będę musiał przekształcić dane w inny format, to wciąż mam oryginalne dane tekstowe, więc mogę po prostu ponownie uruchomić pipeline.

Podział pracy

Czego się nauczyłem: nie próbuj optymalizować zadań ręcznie, niech zajmie się tym komputer.

Debugowałem workflow na jednej chromosomie, teraz trzeba przetworzyć wszystkie pozostałe dane.
Chciałem uruchomić kilka instancji EC2 do konwersji, ale jednocześnie obawiałem się uzyskać skrajnie zbilansowane obciążenie w różnych zadaniach przetwarzania (tak samo jak Spark cierpiał na niezbilansowane partycje). Ponadto, nie uśmiechało mi się uruchamiać jednej instancji na każdą chromosomę, ponieważ dla kont AWS jest domyślne ograniczenie do 10 instancji.

Wtedy postanowiłem napisać skrypt w R do optymalizacji zadań przetwarzania.

Na początku poprosiłem S3 o obliczenie, ile miejsca w pamięci zajmuje każda chromosoma.

biblioteka(aws.s3)
biblioteka(tidyverse)

chr_sizes % 
  mutate(Size = as.numeric(Size)) %>% 
  filter(Size != 0) %>% 
  mutate(
    # Ekstrakcja chromosomu z nazwy pliku 
    chr = str_extract(Key, 'chr.{1,4}.csv') %>%
             str_remove_all('chr|.csv')
  ) %>% 
  group_by(chr) %>% 
  summarise(total_size = sum(Size)/1e+9) # Dzielić, aby uzyskać wartość w GB



# 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.
# … z 17 dodatkowymi wierszami

Następnie napisałem funkcję, która bierze całkowity rozmiar, tasuje kolejność chromosomów, dzieli je na grupy num_jobs i raportuje, jak bardzo różnią się rozmiary wszystkich zadań przetwarzania.

num_jobs <- 7
# Jak duże byłoby każde zadanie, gdyby zostało idealnie podzielone?
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)



# Tibble: 1 x 2
     sd data            
             
1  153.

Potem przetestowałem za pomocą purrr tysiąc tasowań i wybrałem najlepsze.

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

W ten sposób uzyskałem zestaw zadań, które są bardzo podobne pod względem rozmiaru. Potem wystarczyło tylko opakować mój wcześniejszy skrypt Bash w dużą pętlę for. Na napisanie tej optymalizacji zajęło mi około 10 minut. A to znacznie mniej, niż poświęciłbym na ręczne tworzenie zadań w przypadku ich nierównowagi. Dlatego uważam, że z tą wstępną optymalizacją nie popełniłem błędu.

dla DESIRED_CHR w "16" "9" "7" "21" "MT"
do
# Kod do przetwarzania pojedynczego chromosomu
fi

Na końcu dodaję polecenie wyłączenia:

sudo shutdown -h now

… i udało się! Z pomocą AWS CLI uruchomiłem instancje i przez opcję user_data przekazywałem im skrypty Bash ich zadań do przetworzenia. Wykonywały się i automatycznie wyłączały, więc nie płaciłem za nadmiarową moc obliczeniową.

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

Pakujemy!

Czego się nauczyłem: API powinno być proste dla prostoty i elastyczności użycia.

W końcu otrzymałem dane w odpowiednim miejscu i formie. Teraz należało maksymalnie uprościć proces korzystania z danych, aby moim kolegom było łatwiej. Chciałem stworzyć prosty interfejs API do tworzenia zapytań. Jeśli w przyszłości zdecyduję się przejść z .rds jeśli chodzi o pliki Parquet, to powinno to być problemem dla mnie, a nie dla kolegów. Dlatego zdecydowałem się stworzyć wewnętrzny pakiet R.

Zebrałem i udokumentowałem bardzo prosty pakiet, zawierający tylko kilka funkcji do dostępu do danych, zebranych wokół funkcji get_snp. Zrobiłem również stronę dla kolegów pkgdown, aby mogli łatwo zobaczyć przykłady i dokumentację.

Inteligentne buforowanie

Czego się nauczyłem: jeśli Twoje dane są dobrze przygotowane, buforowanie będzie proste!

Ponieważ jeden z głównych workflowów stosował tę samą model analizy dla pakietu SNP, postanowiłem wykorzystać grupowanie (binning) na swoją korzyść. Przy przesyłaniu danych za pomocą SNP do zwracanego obiektu dołączane są również wszystkie informacje z grupy (bin). Oznacza to, że stare zapytania mogą (teoretycznie) przyspieszać przetwarzanie nowych zapytań.

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

Podczas tworzenia pakietu przeprowadziłem wiele benchmarków, aby porównać szybkość stosowania różnych metod. Zalecam, aby nie lekceważyć tego, ponieważ wyniki czasami bywają zaskakujące. Na przykład dplyr::filter okazał się znacznie szybszy niż łapanie wierszy z wykorzystaniem filtracji opartej na indeksowaniu, a pozyskanie jednej kolumny z odfiltrowanego ramki danych działało znacznie szybciej niż zastosowanie składni indeksowania.

Zwróć uwagę, że obiekt prev_snp_results zawiera klucz snps_in_bin. To tablica wszystkich unikalnych SNP w grupie (bin), co pozwala na szybkie sprawdzenie, czy dane z poprzedniego zapytania już istnieją. Ułatwia to również iteracyjne przechodzenie przez wszystkie SNP w grupie (bin) za pomocą tego kodu:

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

Wyniki

Teraz możemy (i zaczęliśmy poważnie) uruchamiać modele i scenariusze, które wcześniej były dla nas niedostępne. Najlepsze jest to, że moi koledzy z laboratorium nie muszą myśleć o żadnych złożonościach. Mają po prostu działającą funkcję.

I chociaż pakiet zwalnia ich od szczegółów, starałem się uczynić format danych na tyle prostym, aby mogli zrozumieć go, jeśli jutro bym nagle zniknął...

Szybkość znacznie wzrosła. Zwykle skanujemy funkcjonalnie znaczące fragmenty genomu. Wcześniej nie mogliśmy tego robić (kosztowało to zbyt wiele), ale teraz, dzięki strukturze grupowej (bin) i buforowaniu, zapytanie o jeden SNP zajmuje średnio mniej niż 0,1 sekundy, a wykorzystanie danych jest tak niskie, że koszty S3 są znikome.

Ostatnio zostałem odpowiedzialny za zarządzanie ponad 25 TB surowych danych genotypowych w moim laboratorium. Kiedy zacząłem, użycie Sparka zajmowało 8 minut i kosztowało 20 USD na zapytanie SNP. Po zastosowaniu AWK + #rstats teraz zajmuje to mniej niż 1/10 sekundy i kosztuje 0,00001 USD. Mój osobisty #BigData sukces. pic.twitter.com/ANOXVGrmkk

— Nick Strayer (@NicholasStrayer) 30 maja 2019

Podsumowanie

Ten artykuł to wcale nie przewodnik. Rozwiązanie okazało się indywidualne i prawdopodobnie nieoptymalne. Raczej to opowieść o podróży. Chcę, aby inni zrozumieli, że podobne rozwiązania nie pojawiają się w głowie w pełni uformowane, to rezultat prób i błędów. Ponadto, jeśli szukasz specjalisty ds. analizy danych, weź pod uwagę, że do efektywnego wykorzystania tych narzędzi potrzebne jest doświadczenie, a doświadczenie wymaga pieniędzy. Cieszę się, że miałem środki na zapłatę, ale wielu innych, którzy mogą wykonać tę samą pracę lepiej ode mnie, nigdy nie będzie miało takiej możliwości z powodu braku pieniędzy na nawet próbę.

Narzędzia do analizy dużych danych są uniwersalne. Jeśli masz czas, prawdopodobnie będziesz w stanie napisać szybsze rozwiązanie, stosując 'inteligentne' czyszczenie danych, składowanie i metody wydobywania. Ostatecznie wszystko sprowadza się do analizy kosztów i korzyści.

Czego się nauczyłem:

• nie ma taniego sposobu na przetworzenie 25 TB na raz;
• uważaj na rozmiar swoich plików Parquet i ich organizację;
• partycje w Sparku powinny być zrównoważone;
• nigdy nie próbuj robić 2,5 miliona partycji;
• sortowanie jest wciąż trudne, tak samo jak konfiguracja Sparka;
• czasami szczególne dane wymagają szczególnych rozwiązań;
• łączenie w Sparku działa szybko, ale partycjonowanie nadal jest kosztowne;
• nie śpij, gdy uczą cię podstaw, na pewno ktoś już rozwiązał twój problem jeszcze w latach 80-tych;
• gnu parallel to magiczna rzecz, wszyscy powinni jej używać;
• Spark lubi nieskompresowane dane i nie lubi łączyć partycji;
• w Sparku koszt energii przy rozwiązywaniu prostych zadań jest zbyt wysoki;
• tablice asocjacyjne w AWK są bardzo wydajne;
• można do nich odwoływać się stdin i stdout z skryptu R, więc można je wykorzystać w potoku;
• dzięki inteligentnej implementacji ścieżek S3 można przetwarzać wiele plików;
• główną przyczyną straty czasu jest przedwczesna optymalizacja metody przechowywania;
• nie próbuj optymalizować zadań ręcznie, niech to robi komputer;
• API powinno być proste ze względu na prostotę i elastyczność użycia;
• jeśli twoje dane są dobrze przygotowane, cachowanie będzie łatwe!

Źródło: habr.com