Как да прочетете тази статия: извинявам се, че текстът стана толкова дълъг и хаотичен. За да спестя вашето време, всяка глава започвам с въведението „Какво научих“, в което в едно или две изречения излагам същността на главата.

«Просто покажи решението!» Ако просто искате да видите до какво стигнах, отидете на главата „Ставам по-иновационен“, но смятам, че е много по-интересно и полезно да прочетете за провалите.

Наскоро ми беше възложено да настроя процеса на обработка на голям обем изходни ДНК последователности (технически това е SNP чип). Трябваше бързо да получа данни за определеното генетично местоположение (което се нарича SNP) за последващо моделиране и други задачи. С помощта на R и AWK успях да почистя и организирам данните по естествен начин, значително ускоривайки обработката на запитвания. Не беше лесно и изискваше многократни итерации. Тази статия ще ви помогне да избегнете някои от моите грешки и ще демонстрира какво всъщност постигнах.

За начало някои въвеждащи обяснения.

Данни

Нашият университетски център за обработка на генетична информация ни предостави данни във формат TSV с обем 25 Тб. Получих ги разделени на 5 пакета, компресирани с Gzip, всеки от които съдържаше около 240 четиригигабайтни файла. Всеки ред съдържаше данни за един SNP на един човек. Общо бяха предадени данни за ~2,5 млн SNP и ~60 хиляди души. Освен информацията за SNP в файловете имаше многобройни колони с числа, отразяващи различни характеристики, като интензивност на четене, честота на различни алели и т.н. Общият брой колони с уникални стойности беше около 30.

Цел

Както в всеки проект за управление на данни, най-важното беше да определим как ще се използват данните. В този случай по-голямата част от времето ще създаваме модели и работни потоци за SNP на базата на SNP. Тоест едновременно ще ни трябват данни само за един SNP. Трябваше да науча как да извличам всички записи, свързани с един от 2,5 милиона SNP, по най-простия, бърз и евтин начин.

Какво да не правим

Ще цитирам подходящо клише:

Не съм се провалял хиляда пъти, просто открих хиляда начина да не парсвам куп данни в удобен за запитвания формат.

Първи опит.

Какво научих: не съществува евтин начин да се парсират 25 Тб наведнъж.

След като слушах курса „Разширени методи за обработка на големи данни“ в Университета Вандербилт, бях уверен, че всичко е наред. Вероятно ще отнеме един или два часа настройването на Hive-сървър, за да премина през всички данни и да отчета резултатите. Тъй като нашите данни се съхраняват в AWS S3, аз използвах услугата Athena, която позволява прилагането на Hive SQL заявки към S3 данните. Не е нужно да настройвате/изграждате Hive клъстер и плащате само за данните, които търсите.

След като показах на Athena моите данни и техния формат, изпълних няколко теста с подобни заявки:

select * from intensityData limit 10;

И бързо получих добре структурирани резултати. Готово.

Докато не опитахме да използваме данните в работа...

Помолиха ме да изтегля цялата информация за SNP, за да тествам модела. Изпълних заявката:

select * from intensityData 
where snp = 'rs123456';

…и започнах да чакам. След осем минути и повече от 4 Тб запитани данни, получих резултата. Athena начислява такса за обема на намерените данни, по 5 $ за терабайт. Така че, това единствено запитване струваше 20 $ и осем минути чакане. За да пусна модела върху всички данни, трябваше да чакам 38 години и да платя 50 млн. $. Очевидно, че това не беше подходящо за нас.

Трябваше да използваме Parquet...

Какво научих: бъдете внимателни с размера на вашите Parquet файлове и тяхната организация.

Първоначално се опитах да поправя ситуацията, като конвертирах всички TSV в Parquet файлове. Те са удобни за работа с големи набори от данни, защото информацията в тях се съхранява в колонен формат: всяка колона лежи в собствен сегмент памет/диск, за разлика от текстовите файлове, където редовете съдържат елементите на всяка колона. И ако трябва нещо да се намери, достатъчно е да прочетете необходимата колона. Освен това, всеки файл съдържа диапазон от стойности в колоната, така че ако търсената стойност липсва в диапазона на колоната, Spark няма да губи време да сканира целия файл.

Изпълних проста задача AWS Glue за преобразуването на нашите TSV в Parquet и си пуснах нови файлове в Athena. Отне около 5 часа. Но когато стартирах запитването, за изпълнението му отиде приблизително същото време и малко по-малко пари. Работата е там, че Spark, опитвайки се да оптимизира задачата, просто разопакова един TSV-чанк и го положи в собствен Parquet-чанк. И тъй като всеки чанк беше достатъчно голям и съдържаше пълни записи на много хора, то във всеки файл имаше всички SNP, затова на Spark му се наложи да отвори всички файлове, за да извлече необходимата информация.

Любопитно е, че използваният по подразбиране (и препоръчителен) тип компресия в Parquet — snappy — не е разделяем (splitable). Затова всеки изпълнител (executor) заседна на задачата по разопаковане и зареждане на целия датасет от 3,5 Гб.

Разглеждаме проблема

Какво научих: трудно е да се сортира, особено когато данните са разпределени.

Мислех, че сега разбирам същността на проблема. Нужно ми беше само да сортирам данните по колоната SNP, а не по хора. Тогава в отделен чанк данни ще се съхраняват няколко SNP и тогава "умната" функция на Parquet "отваря само ако стойността е в диапазон" ще се прояви в цялата си слава. За съжаление, да се сортират милиарди редове, разпръснати из клъстера, се оказа сложна задача.

Аз, взимайки курс по алгоритми в колежа: «Уф, никой не се интересува от изчислителната сложност на всички тези сортировъчни алгоритми»

Аз, опитвайки се да сортирам по колона в 20TB #spark таблица: «Защо това отнема толкова време?» #DataScience трудности.

— Nick Strayer (@NicholasStrayer) 11 март 2019

AWS определено не иска да върне парите по причината «Аз съм разсеян студент». След като стартирах сортирането на Amazon Glue, то работи 2 дни и завърши с провал.

Какво ще кажете за партиционирането?

Какво научих: партициите в Spark трябва да бъдат балансирани.

Тогава ми хрумна идеята да партиционирам данните по хромозомите. Има 23 от тях (и още няколко, ако вземем предвид митохондриалната ДНК и неразшифрованите области).
Това ще позволи разделяне на данните на по-малки порции. Ако добавим в Spark функция за експортиране в скрипта Glue само една редица partition_by = "chr", данните трябва да бъдат разпределени по бакети (buckets).

Геномът се състои от многобройни фрагменти, наречени хромозоми.

За съжаление, това не проработи. Хромозомите имат различни размери, а следователно и различно количество информация. Това означава, че задачите, които Spark изпращаше на работниците, не бяха балансирани и се изпълняваха бавно, тъй като някои възли завършваха по-рано и просто стояха без работа. Въпреки това задачите бяха изпълнени. Но когато поисках един SNP, небалансираността отново стана причина за проблеми. Цената за обработка на SNP в по-големите хромозоми (т.е. от там, откъдето искаме да получим данни) намаля само с около 10 пъти. Много, но не достатъчно.

А какво ако разделим на още по-малки партиции?

Какво научих: никога не се опитвайте да създавате 2,5 милиона партиции.

Реших да се разходя и партиционирах всеки SNP. Това гарантираше еднакъв размер на партициите. БЕШЕ ЛОША ИДЕЯ. Използвах Glue и добавих невинна редица 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')
  )

…въпреки това задачата все пак не беше завършена. Настройвах по всевъзможен начин: увеличавах паметта за всеки изпълнител на запитвания, използвах възли с по-голям капацитет на паметта, прилагах широковещателни променливи (broadcasting variable), но всеки път това се оказваше половинчато решение и постепенно изпълнителите започваха да се провалят, докато всичко не спря.

Актуализация: така започва. pic.twitter.com/agY4GU2ru5

— Nick Strayer (@NicholasStrayer) 15 май 2019

Становя по-изобретателен

Какво научих: понякога специализирани данни изискват специализирани решения.

Всеки SNP има стойност на позиция. Това е число, което отговаря на броя на основите, разположени по неговата хромозома. Това е добър и естествен начин за организиране на нашите данни. Първоначално исках да направя партициониране по области на всяка хромозома. Например, позиции 1 - 2000, 2001 - 4000 и т.н. Но проблемът е, че SNP е неравномерно разпределен по хромозомите, така че размерът на групите ще варира значително.

В резултат на това стигнах до разбивка по категории (ранг) на позициите. По вече заредените данни извърших запитване за получаване на списък с уникални SNP, техните позиции и хромозоми. След това сортирах данните в рамките на всяка хромозома и събрах SNP в групи (бин) с определен размер. Да кажем, по 1000 SNP. Това ми даде връзката SNP с групата-в-хромозома.

В крайна сметка направих групи (бин) по 75 SNP, причината ще обясня по-долу.

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

Първият опит със Spark

Какво научих: обединението в 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

Какво научих: не спете, когато ви преподават основите. Сигурно някой вече е решил вашия проблем още през 80-те години.

Досега причината за всичките ми неуспехи със 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 Мб памет. За да реша проблема и да не загубя куп пари, трябваше да измисля как да разпаралелета работата. За щастие, в изключително страхотната книга Data Science at the Command Line на Джерон Джансенс открих глава, посветена на разпаралеляването. От нея научих за gnu parallel, много гъвкав метод за реализиране на многопоточност в 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 терабайта пакет само за няколко часа.

Няма нищо по-сладко от това да видиш как всички ядра, за които плащаш в AWS, се използват. Благодарение на gnu-parallel мога да разархивирам и разделя CSV файл от 19 гигабайта толкова бързо, колкото мога да го изтегля. Дори не можах да накарам Spark да работи. #DataScience #Linux pic.twitter.com/Nqyba2zqEk

— Nick Strayer (@NicholasStrayer) 17 май 2019

Този туит е трябвало да спомене 'TSV'. За съжаление.

Повторно използване на парснати данни

Какво научих: Spark харесва несжати данни и не обича да комбинира партиции.

Сега данните бяха в S3 в несжатиран (с разделители) и полуподреден формат, и можех отново да се върна към Spark. Очакваше ме изненада: отново не успях да постигна желаното! Беше много трудно точно да кажа на Spark как са партиционирани данните. И дори когато го направих, се оказа, че партициите са твърде много (95 000) и когато с помощта на coalesce намалих количеството им до разумни граници, това разруши моето партициониране. Сигурно може да се поправи, но след няколко дни търсене не успях да намеря решение. В крайна сметка успях да завърша всички задачи в Spark, макар че отне известно време, а моите разделени Parquet файлове не бяха много малки (~200 Кб). Въпреки това, данните се намираха там, където трябва.

Твърде малки и неравномерни, чудесно!

Тестове на локални Spark заявки

Какво научих: в Spark има твърде много разходи при решаване на прости задачи.

Като заредих данните в обмислен формат, успях да тествам скоростта. Настроих скрипт на R за стартиране на локален Spark сървър и после заредих Spark DataFrame от посоченото хранилище на Parquet групи (bin). Опитвах се да заредя всичките данни, но не можах да накарам Sparklyr да разпознае партиционирането.

sc <- Spark_connect(master = "local")

desired_snp <- 'rs34771739'

# Започнете таймер
start_time <- Sys.time()

# Заредете желания bin в Spark
intensity_data % 
  Spark_read_Parquet(
    name = 'intensity_data', 
    path = get_snp_location(desired_snp),
    memory = FALSE )

# Подмножество bin за snp и след това събиране на локално
test_subset % 
  filter(SNP_Name == desired_snp) %>% 
  collect()

print(Sys.time() - start_time)

Изпълнението отне 29,415 секунди. Много по-добре, но не достатъчно добро за масово тестване на нещо. Освен това не можах да увелича скоростта чрез кеширане, защото когато се опитвах да кеширам в паметта DataFrame, Spark винаги се сриваше, дори когато отделях повече от 50 Гб памет за датасета, който тежеше по-малко от 15.

Връщане към AWK

Какво научих: асоциативните масиви в AWK са много ефективни.

Разбрах, че мога да постигна по-висока скорост. Спомних си, че в забележителното ръководство за AWK на Брус Барнет Четох за готина функция, наречена «асоциативни масиви». Всъщност, това са двойки ключ-стойност, които по някаква причина в AWK са наречени по друг начин и затова не съм си спомнял особено за тях. Роман Чепляка ми напомни, че терминът «асоциативни масиви» е много по-стар от термина «пара ключ-стойност». Дори ако потърсите ключ-стойност в Google Ngram, този термин няма да го видите там, но ще намерите асоциативни масиви! Освен това «пара ключ-стойност» обикновено асоциира с бази данни, така че е много по-логично да се сравнява с hashmap. Разбрах, че мога да използвам тези асоциативни масиви, за да свържа моите SNP с таблицата на групите (bin table) и суровите данни без използване на Spark.

За това в AWK скрипта използвах блока BEGIN. Това е фрагмент от кода, който се изпълнява преди първият ред данни да бъде предаден на основното тяло на скрипта.

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

Команда while(getline...) зареди всички редове от CSV групата (bin), зададе първата колона (имената на SNP) като ключ за асоциативния масив bin и втората стойност (групата) като стойност. След това в блока { }, който се изпълнява за всички редове на основния файл, всеки ред се изпраща в изходния файл, който получава уникално име в зависимост от неговата група (bin): ..._bin_"bin[$1]"_....

Променливи batch_num и chunk_id съответстваха на данните, предоставени от конвейера, което позволи да се избегне състояние на гонка, и всеки поток на изпълнение, стартиран паралелно, записваше в собствен уникален файл.

Тъй като всички сурови данни разпределих по папки по хромозоми, останали след предишния ми експеримент с AWK, сега можех да напиша друг Bash скрипт, за да обработвам по хромозома наведнъж и да качвам в S3 по-дълбоко партиционирани данни.

DESIRED_CHR='13'

# Изтегляне на данни за хромозомите от s3 и разделяне на чанкове
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"

# Комбиниране на всички чанкове от паралелния процес в единични файлове и качване в RDS с помощта на 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/*

В скрипта има два раздела паралелно.

В първия раздел се четат данни от всички файлове, съдържащи информация за необходимата хромозома, след което тези данни се разпределят по потоци, които разпределят файловете в съответните групи (bin). За да не се получи състояние на конкуренция, когато няколко потока записват в един файл, AWK предава имената на файловете за запис на данни на различни места, например, chr_10_bin_52_batch_2_aa.csv. В резултат на това на диска се създават множество малки файлове (за това използвах терабайтни EBS томове).

Конвейерът от втория раздел паралелно преминава през групите (bin) и обединява тяхните отделни файлове в общи CSV с cat, а след това ги изпраща за експортиране.

Транслиране в R?

Какво научих: можете да се обръщате към stdin и stdout от R скрипта, което означава, че можете да го използвате в конвейера.

В Bash скрипта можете да забележите следния ред: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Той транслира всички конкатенирани файлове на групата (bin) в по-долу представения R скрипт. {} е специална техника паралелно, която всякакви данни, изпратени от нея в указания поток, вкарва директно в самата команда. Опцията {#} предоставя уникален ID на изпълнителния поток, а {%} представлява номер на слота на задачата (повтаря се, но никога едновременно). Списък на всички опции може да бъде намерен в документацията.

#!/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 е нещо като по-умалена версия на Parquet, без изтънченостите на колоночното хранилище.

След завършване на Bash скрипта получих купчина .rds- файлове, разположени в S3, което ми позволи да използвам ефективно компресиране и вградени типове.

Въпреки използването на бавен R, всичко работеше много бързо. Не е учудващо, че фрагментите на R, отговарящи за четене и запис на данни, са добре оптимизирани. След тестване на една хромозома със среден размер, задачата беше изпълнена на инстанция C5n.4xl за около два часа.

Ограничения на S3

Какво научих: благодарение на интелигентната реализация на пътищата S3 може да обработва много файлове.

Притеснявах се, дали S3 ще може да обработи множеството предадени й файлове. Можех да направя имената на файловете осмислени, но как S3 ще търси по тях?

Папките в S3 са просто за красота, на самата система не й интересува символ. /. От FAQ страницата на S3.

Изглежда, че S3 представлява път към конкретен файл като прост ключ в своеобразна хеш-таблица или база данни на основата на документи. Бакетът (bucket) може да се счита за таблица, а файловете — за записи в тази таблица.

Тъй като скоростта и ефективността са важни за получаване на печалба в Amazon, не е изненадващо, че тази система „ключ-в-качеството-на-пътя-до-файла“ е невероятно оптимизирана. Опитвах се да намеря баланс: да няма нужда от множество get-запроси, но също така да се изпълняват бързо. Оказа се, че най-добре е да се създадат около 20 000 bin-файла. Мисля, че ако продължа да оптимизирам, може да се постигне увеличение на скоростта (например, да се създаде специален бакет само за данни, което да намали размера на поисковата таблица). Но нямах време и средства за по-нататъшни експерименти.

Какво ще кажете за кръстосаната съвместимост?

Какво научих: основната причина за загуба на време е преждевременната оптимизация на метода за съхранение.

В този момент е много важно да се запитате: „Защо да използвам собствен формат на файлове?“ Причината е в скоростта на зареждане (опакованите gzip CSV-файлове се зареждаха 7 пъти по-бавно) и съвместимостта с нашите работни процеси. Мога да преразгледам решението си, ако R може лесно да зарежда файлове Parquet (или Arrow) без натоварване от Spark. В лабораторията всички използват R и, ако трябва да преобразувам данните в друг формат, все още разполагам с оригиналните текстови данни, така че мога просто да стартирам конвейера отново.

Разпределение на работата

Какво научих: не се опитвайте да оптимизирате задачите ръчно, оставете компютъра да го направи.

Настроих работния процес на една хромозома, сега трябва да обработя всичките останали данни.
Исках да стартирам няколко инстанса EC2 за преобразуване, но същевременно се опасявах да не получа крайно небалансирано натоварване при различните задачи по обработка (както и Spark страдаше от небалансирани партиции). Освен това не исках да стартирам по един инстанс за всяка хромозома, защото за AWS акаунтите има ограничение по подразбиране от 10 инстанса.

Тогава реших да напиша на R скрипт за оптимизиране на задачите по обработка.

Първо помолих S3 да изчисли колко място в хранилището заема всяка хромозома.

library(aws.s3)
library(tidyverse)

chr_sizes % 
  mutate(Size = as.numeric(Size)) %>% 
  filter(Size != 0) %>% 
  mutate(
    # Извлечете хромозом от името на файла 
    chr = str_extract(Key, 'chr.{1,4}.csv') %>%
             str_remove_all('chr|.csv')
  ) %>% 
  group_by(chr) %>% 
  summarise(total_size = sum(Size)/1e+9) # Делим, за да получим стойността в GB



# А тибли: 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.
# … с още 17 реда

След това написах функция, която взема общия размер, разбърква реда на хромозомите и ги разделя на групи num_jobs и съобщава, колко се различават размерите на всички задания за обработка.

num_jobs <- 7
# Колко голямо би било всяко задание, ако е перфектно разделено?
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)



# А тибли: 1 x 2
     sd data            
   <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
# Код за обработка на един хромозом
fi

В края добавям команда за изключване:

sudo shutdown -h now

… и всичко стана! С помощта на AWS CLI създавах инстанции и чрез опцията user_data предавах им Bash-скриптове за тяхните задания за обработка. Те се изпълняваха и автоматично се изключваха, така че не плащах за излишна изчислителна мощност.

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

Опаковаме!

Какво научих: API-ят трябва да е прост заради простотата и гъвкавостта на използването.

Накрая получих данните на правилното място и в правилния вид. Оставаше само да опростя максимално процеса на използване на данните, за да е по-лесно за моите колеги. Исках да направя прост API за създаване на заявки. Ако в бъдеще реша да премина към .rds Ако говорим за Parquet файлове, това трябва да е проблем за мен, а не за колегите. Затова реших да създам вътрешен R пакет.

Събрах и документирах много прост пакет, който съдържа само няколко функции за достъп до данните, агрегирани около функцията get_snp. Създадох и уебсайт за колегите, pkgdown, за да могат да преглеждат примери и документация с лекота.

Интелигентно кеширане

Какво научих: ако данните ви са добре подготвени, кеширането ще е лесно!

Тъй като един от основните работни потоци прилага същия анализен модел към SNP пакета, реших да използвам групиране (биндинг) в своя полза. Когато прехвърляте данни по SNP, към върнатия обект се прикрепя и цялата информация от групата (бин). Тоест старите заявки могат (теоретично) да ускорят обработката на нови заявки.

# 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 в групата (бин), който позволява бърза проверка за наличието на данни от предходната заявка. Освен това улеснява цикличното обхождане на всички SNP в групата (бин) с помощта на следния код:

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

Резултати

Сега можем (и започнахме сериозно) да прогонваме модели и сценарии, които преди бяха недостъпни за нас. Най-доброто е, че на колегите ми в лабораторията не им се налага да мислят за всички сложности. Просто имаме работеща функция.

И макар пакетът да ги избавя от детайлите, се опитах да направя формата на данните достатъчно прост, за да могат да се справят с него, ако утре изведнъж изчезна …

Скоростта значително се е увеличила. Обикновено сканираме функционално значими фрагменти от генома. Преди не можехме да го правим (беше твърде скъпо), но сега, благодарение на груповата (бин) структура и кеширането, за заявка на един SNP отнема средно по-малко от 0.1 секунди, а използването на данни е толкова ниско, че разходите за получаване на S3 са символични.

Наскоро бях назначен да се занимавам с обработката на над 25 ТБ сурови генотипни данни за моята лаборатория. Когато започнах, използването на Spark отнемаше 8 минути и струваше 20 $ за запитване на SNP. След като използвах AWK + #rstats за обработка, сега отнема по-малко от десета от секундата и струва 0.00001 $. Личната ми #BigData победа. pic.twitter.com/ANOXVGrmkk

— Nick Strayer (@NicholasStrayer) 30 май 2019 г.

Заключение

Тази статия не е ръководство. Решението е индивидуално и почти сигурно не оптимално. По-скоро това е разказ за пътуването. Искам другите да разберат, че подобни решения не идват напълно оформени в главата, а са резултат от опити и грешки. Освен това, ако търсите специалист по анализ на данни, имайте предвид, че за ефективното използване на тези инструменти е нужен опит, а опитът изисква пари. Радвам се, че имах средства за заплата, но много други, които могат да свършат същата работа по-добре от мен, никога няма да имат такава възможност поради липсата на средства дори за опит.

Инструментите за големи данни са универсални. Ако имате време, почти сигурно можете да напишете по-бързо решение, прилагайки "умна" очистка на данните, хранилище и методи за извличане. В крайна сметка всичко опира до анализа на разходите и ползите.

Научих, че:

• няма евтин начин да се обработят 25 ТБ наведнъж;
• внимавайте с размера на вашите Parquet файлове и тяхната организация;
• партиции в Spark трябва да бъдат балансирани;
• никога не се опитвайте да правите 2,5 милиона партиции;
• сортировката все още е трудна, както и настройката на Spark;
• понякога специални данни изискват специални решения;
• обединението в Spark работи бързо, но партиционирането все още излиза скъпо;
• не спите, когато ви преподават основи — със сигурност някой вече е решил проблема ви още през 80-те;
• gnu parallel — това е магическа вещ, всички трябва да я използват;
• Spark обича несжати данни и не обича да комбинира партиции;
• в Spark има твърде много разходи при решаването на прости задачи;
• асоциативните масиви в AWK са много ефективни;
• можете да се обадите на stdin и stdout от R скрипт, което означава, че можете да го използвате в конвейера;
• благодарение на умната реализация на пътища S3 може да обработва много файлове;
• главната причина за загуба на време е преждевременната оптимизация на метода на съхранение;
• не се опитвайте да оптимизирате заданията ръчно, оставете компютъра да го прави;
• API трябва да бъде прост заради простота и гъвкавост при използването;
• ако данните ви са добре подготвени, кэширане ще бъде просто!

Източник: habr.com