Како да ја прочитате оваа статија: Се извинувам што текстот е толку долг и хаотичен. За да ви заштеди време, секое поглавје го започнувам со вовед „Што научив“, кој ја сумира суштината на поглавјето во една или две реченици.

„Само покажи ми го решението!“ Ако само сакате да видите од каде дојдов, тогаш прескокнете го поглавјето „Да станете поинвентивни“, но мислам дека е поинтересно и покорисно да се чита за неуспехот.

Неодамна добив задача да поставам процес за обработка на голем обем на сурови ДНК секвенци (технички SNP чип). Потребата беше брзо да се добијат податоци за дадена генетска локација (наречена SNP) за последователно моделирање и други задачи. Користејќи ги R и AWK, можев да ги исчистам и организирам податоците на природен начин, со што значително ја забрзав обработката на барањата. Ова не ми беше лесно и бараше бројни повторувања. Оваа статија ќе ви помогне да избегнете некои од моите грешки и да ви покажеме со што завршив.

Прво, некои воведни објаснувања.

Податоци

Нашиот универзитетски центар за обработка на генетски информации ни обезбеди податоци во форма на TSV од 25 TB. Ги добив поделени во 5 Gzip-компресирани пакети, од кои секој содржи околу 240 датотеки од четири гигабајти. Секој ред содржеше податоци за еден SNP од еден поединец. Вкупно беа пренесени податоци за ~ 2,5 милиони SNP и ~ 60 илјади луѓе. Покрај информациите за SNP, датотеките содржеа бројни колони со броеви што одразуваат различни карактеристики, како што се интензитетот на читање, фреквенцијата на различни алели итн. Вкупно имаше околу 30 колони со единствени вредности.

цел

Како и со секој проект за управување со податоци, најважно беше да се одреди како ќе се користат податоците. Во овој случај ние најмногу ќе избираме модели и работни текови за SNP врз основа на SNP. То есть одновременно нам будут нужны данные только по одному SNP. Я должен был научиться как можно проще, быстрее и дешевле извлекать все записи, относящиеся к одному из 2,5 миллионов SNP.

Како да не го направите ова

Да цитирам соодветно клише:

Не успеав илјада пати, само открив илјада начини да избегнам парсирање на куп податоци во формат погоден за пребарување.

Прво пробајте

Што научив: Не постои евтин начин да се анализираат 25 ТБ истовремено.

Откако го поминав курсот „Напредни методи за обработка на големи податоци“ на Универзитетот Вандербилт, бев сигурен дека трикот е во торбата. Веројатно ќе бидат потребни еден или два часа за да се постави серверот Hive да работи низ сите податоци и да го пријави резултатот. Бидејќи нашите податоци се зачувани во AWS S3, ја користев услугата Атена, што ви овозможува да ги примените барањата Hive SQL на податоците на S3. Не треба да поставувате/подигате кластер Hive, а исто така плаќате само за податоците што ги барате.

Откако и ги покажав на Атина моите податоци и неговиот формат, извршив неколку тестови со прашања како ова:

select * from intensityData limit 10;

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

Се додека не се обидовме да ги искористиме податоците во нашата работа ...

Од мене беше побарано да ги извадам сите информации за SNP за да го тестирам моделот. Го извршив барањето:

select * from intensityData 
where snp = 'rs123456';

...и почна да чека. По осум минути и повеќе од 4 ТБ барани податоци, го добив резултатот. Атина наплаќа според обемот на пронајдени податоци, 5 долари по терабајт. Така, ова единствено барање чинеше 20 долари и осум минути чекање. За да го вклучиме моделот на сите податоци, моравме да чекаме 38 години и да платиме 50 милиони долари.Очигледно, ова не ни одговараше.

Беше неопходно да се користи Паркет...

Што научив: Бидете внимателни со големината на вашите Паркет датотеки и нивната организација.

Прво се обидов да ја поправам ситуацијата со конвертирање на сите TSV во Паркетни датотеки. Тие се погодни за работа со големи збирки податоци бидејќи информациите во нив се складираат во колонообразна форма: секоја колона лежи во сопствената меморија/сегмент на дискот, за разлика од текстуалните датотеки, во кои редовите содржат елементи од секоја колона. И ако треба да најдете нешто, тогаш само прочитајте ја потребната колона. Дополнително, секоја датотека складира опсег на вредности во колона, па ако вредноста што ја барате не е во опсегот на колоната, Spark нема да губи време скенирање на целата датотека.

Истрчав едноставна задача AWS лепак да ги конвертираме нашите TSV во паркет и да ги исфрлиме новите датотеки во Athena. Беа потребни околу 5 часа. Но, кога го извршив барањето, ми требаше исто толку време и малку помалку пари за да се пополни. Факт е дека Spark, обидувајќи се да ја оптимизира задачата, едноставно отпакува едно парче TSV и го стави во сопственото парче Паркет. И бидејќи секое парче беше доволно големо за да ги собере целосните записи на многу луѓе, секоја датотека ги содржеше сите SNP, па Spark мораше да ги отвори сите датотеки за да ги извлече потребните информации.

Интересно е што стандардниот (и препорачан) тип на компресија на Parquet, snappy, не може да се подели. Затоа, секој извршител беше заглавен во задачата да ја распакува и преземе целата база на податоци од 3,5 GB.

Ајде да го разбереме проблемот

Што научив: Сортирањето е тешко, особено ако податоците се дистрибуираат.

Мне казалось, что теперь я понял суть проблемы. Мне нужно было лишь отсортировать данные по колонке SNP, а не по людям. Тогда в отдельном чанке данных будет храниться несколько SNP, и тогда себя во всей красе проявит «умная» функция Parquet «открывать только если значение находится в диапазоне». К сожалению, отсортировать миллиарды строк, раскиданных по кластеру, оказалось сложной задачей.

Јас земам часови по алгоритми на колеџ: „Уф, никој не се грижи за пресметковната сложеност на сите овие алгоритми за сортирање“

Се обидувам да подредам колона во 20 ТБ #искра табела: „Зошто трае толку долго?“ # Наука за податоци борби.

— Ник Страјер (@NicholasStrayer) Март 11, 2019

AWS дефинитивно не сака да издаде поврат на пари поради причината „Јас сум расеан студент“. Откако трчав со сортирање на Амазон лепак, тој работеше 2 дена и се урна.

Што е со партиционирањето?

Што научив: партиции в Spark должны быть сбалансированы.

Потоа дојдов до идеја за поделба на податоците во хромозомите. Има 23 од нив (и уште неколку ако се земат предвид митохондријалната ДНК и немапираните региони).
Ова ќе ви овозможи да ги поделите податоците на помали парчиња. Ако додадете само една линија на функцијата за извоз Spark во скриптата Glue partition_by = "chr", тогаш податоците треба да се поделат на кофи.

Геномот се состои од бројни фрагменти наречени хромозоми.

За жал, тоа не функционираше. Хромозомите имаат различни големини, што значи различни количини на информации. Ова значи дека задачите што Spark им ги испраќаше на работниците не беа избалансирани и полека завршуваа бидејќи некои јазли завршија рано и беа неактивен. Сепак, задачите беа завршени. Но, кога се бараше една СНП, нерамнотежата повторно предизвика проблеми. Трошоците за обработка на SNP на поголемите хромозоми (т.е. каде сакаме да добиеме податоци) се намалија само за околу 10 фактор. Многу, но не доволно.

А если разделить на ещё более мелкие партиции?

Што научив: Никогаш не обидувајте се да направите 2,5 милиони партиции воопшто.

Решив да одам до крај и го поделив секој SNP. Ова осигури дека партициите се со еднаква големина. БЕШЕ ЛОША ИДЕЈА. Користев Лепак и додадов невина линија partition_by = 'snp'. Задачата започна и почна да се извршува. Еден ден подоцна проверив и видов дека сè уште нема ништо напишано на S3, па ја убив задачата. Изгледа дека Glue пишувал средни датотеки на скриена локација во S3, многу датотеки, можеби неколку милиони. Како резултат на тоа, мојата грешка чинеше повеќе од илјада долари и не му угоди на мојот ментор.

Поделба + сортирање

Што научив: Сортирањето е сè уште тешко, како и подесувањето на Spark.

Мојот последен обид за партиција вклучуваше да ги поделам хромозомите и потоа да ја подредам секоја партиција. Теоретски, ова би го забрзало секое барање бидејќи саканите SNP податоци морале да бидат во рамките на неколку Паркет парчиња во даден опсег. За жал, сортирањето дури и поделени податоци се покажа како тешка задача. Како резултат на тоа, се префрлив на 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')
  )

...сепак, задачата сè уште не беше завршена. Го конфигурирав на различни начини: ја зголемив распределбата на меморијата за секој извршител на барање, користев јазли со голема количина меморија, користев променливи за емитување (променливи за емитување), но секој пат кога тие се покажаа како полумерки и постепено извршителите почнуваа да пропадне додека сè не престане.

Update: so it begins. pic.twitter.com/agY4GU2ru5

— Ник Страјер (@NicholasStrayer) Може 15, 2019

Станувам покреативен

Што научив: Понекогаш специјалните податоци бараат посебни решенија.

Секој SNP има вредност на позицијата. Ова е број кој одговара на бројот на бази долж неговиот хромозом. Ова е убав и природен начин да ги организираме нашите податоци. Отпрвин сакав да поделам по региони на секој хромозом. На пример, позиции 1 - 2000, 2001 - 4000, итн. Но, проблемот е што SNP-ите не се рамномерно распоредени низ хромозомите, така што големината на групата ќе се разликува многу.

Како резултат на тоа, дојдов до расчленување на позициите во категории (ранг). Користејќи ги веќе преземените податоци, побарав да добијам листа на единствени SNP, нивните позиции и хромозоми. Потоа ги подредив податоците во секој хромозом и собрав SNP во групи (канта) со дадена големина. Да речеме по 1000 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 меморија. За да го решиме проблемот и да не изгубиме многу пари, моравме да сфатиме како да ја паралелизираме работата. За среќа, во апсолутно неверојатна книга 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 гига, исто толку брзо колку што можам да го преземам. Не можев да добијам ниту искра за да го водам ова. # Наука за податоци # Линукс pic.twitter.com/Nqyba2zqEk

— Ник Страјер (@NicholasStrayer) Може 17, 2019

Овој твит требаше да го спомене „TSV“. За жал.

Користење на новопаризирани податоци

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

Сега податоците беа во S3 во неотпакуван (читај: споделен) и полуорганизиран формат и можев повторно да се вратам во Spark. Ме чекаше изненадување: Повторно не успеав да го постигнам тоа што го сакав! Беше многу тешко да се каже на Spark како точно се поделени податоците. И дури и кога го направив ова, се покажа дека има премногу партиции (95 илјади), и кога користев coalesce го намали нивниот број на разумни граници, ова ја уништи мојата партиција. Сигурен сум дека ова може да се поправи, но по неколку дена пребарување не можев да најдам решение. На крајот ги завршив сите задачи во Spark, иако ми требаше некое време и моите поделени датотеки на Parquet не беа многу мали (~200 KB). Сепак, податоците беа таму каде што беа потребни.

Премалку и нерамномерно, прекрасно!

Тестирање на локални прашања на 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 од Брус Барнет Прочитав за кул функција наречена „асоцијативни низи" Во суштина, ова се парови со клучна вредност, кои поради некоја причина се нарекуваа поинаку во AWK, и затоа некако не размислував многу за нив. Роман Чепљака потсети дека терминот „асоцијативни низи“ е многу постар од терминот „пар клуч-вредност“. Дури и ако вие побарајте ја вредноста на клучот во Google Ngram, нема да го видите овој термин таму, но ќе најдете асоцијативни низи! Покрај тоа, „парот клуч-вредност“ најчесто се поврзува со бази на податоци, така што има многу пологично да се споредува со хашмапа. Сфатив дека можам да ги користам овие асоцијативни низи за да ги поврзам моите SNP со табела за отпадоци и необработени податоци без употреба на 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 (бин), постави ја првата колона (име на SNP) како клуч за асоцијативната низа bin а втората вредност (група) како вредност. Потоа во блокот { }, кој се извршува на сите линии од главната датотека, секоја линија се испраќа до излезната датотека, која добива единствено име во зависност од нејзината група (бин): ..._bin_"bin[$1]"_....

Променливи batch_num и chunk_id се совпаѓаат со податоците обезбедени од гасоводот, избегнувајќи состојба на трка и секоја нишка за извршување работи 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.

Во првиот дел, податоците се читаат од сите датотеки што содржат информации за саканиот хромозом, потоа овие податоци се дистрибуираат меѓу нишките, кои ги дистрибуираат датотеките во соодветните групи (бин). За да се избегнат услови за трка кога повеќе нишки пишуваат во иста датотека, AWK ги пренесува имињата на датотеките за да запише податоци на различни места, на пр. chr_10_bin_52_batch_2_aa.csv. Како резултат на тоа, многу мали датотеки се креираат на дискот (за ова користев терабајтни тома EBS).

Транспортер од вториот дел parallel поминува низ групите (бин) и ги комбинира нивните индивидуални датотеки во заеднички CSV c catа потоа ги испраќа на извоз.

Емитување во Р?

Што научив: Можете да контактирате stdin и stdout од скрипта R, и затоа користете ја во нафтоводот.

Можеби сте ја забележале оваа линија во вашата скрипта Bash: ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R.... Ги преведува сите поврзани групни датотеки (бин) во скриптата R подолу. {} е посебна техника parallel, кој ги вметнува сите податоци што ги испраќа до наведениот поток директно во самата команда. Опција {#} обезбедува единствен 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 се само за прикажување, всушност системот не го интересира симболот /. Од страницата ЧПП за S3.

Се чини дека S3 ја претставува патеката до одредена датотека како едноставен клуч во некој вид хеш табела или база на податоци базирана на документи. Кофата може да се смета како табела, а датотеките може да се сметаат за записи во таа табела.

Бидејќи брзината и ефикасноста се важни за да се оствари профит во Амазон, не е изненадување што овој систем за патека на клучот како датотека е неверојатно оптимизиран. Се обидов да најдам рамнотежа: за да не морам да поднесувам многу барања за добивање, туку барањата да се извршуваат брзо. Се испостави дека најдобро е да се направат околу 20 илјади датотеки за канти. Мислам дека ако продолжиме да се оптимизираме, можеме да постигнеме зголемување на брзината (на пример, да направиме специјална корпа само за податоци, а со тоа да ја намалиме големината на табелата за пребарување). Но, немаше време или пари за понатамошни експерименти.

Што е со вкрстената компатибилност?

Што научив: причина број еден за потрошено време е предвремено оптимизирање на начинот на складирање.

Во овој момент, многу е важно да се запрашате: „Зошто да користите комерцијален формат на датотека? Причината лежи во брзината на вчитување (вчитувањето на CSV-датотеките на gzipped беше потребно 7 пати подолго) и компатибилноста со нашите работни текови. Може повторно да размислам дали R може лесно да вчита датотеки со Parquet (или стрелка) без оптоварување Spark. Сите во нашата лабораторија користат 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

На крајот ја додавам командата за исклучување:

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 на Паркет фајлови, тогаш ова треба да ми е проблем, а не на моите колеги. За ова решив да направам внатрешен 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+ ТБ необработени податоци за генотипирање за мојата лабораторија. Кога почнав, користењето на искра траеше 8 минути и чинеше 20 долари за барање SNP. По користењето на AWK+ #rstats to process, it now takes less than a 10th of a second and costs $0.00001. My personal #Голем податок победи. pic.twitter.com/ANOXVGrmkk

— Ник Страјер (@NicholasStrayer) Може 30, 2019

Заклучок

Оваа статија воопшто не е водич. Решението се покажа како индивидуално, и речиси сигурно не е оптимално. Напротив, тоа е патопис. Сакам другите да разберат дека таквите одлуки не изгледаат целосно формирани во главата, тие се резултат на обиди и грешки. Исто така, ако барате научник за податоци, имајте на ум дека ефективно користење на овие алатки бара искуство, а искуството чини пари. Среќен сум што имав средства да платам, но многу други кои можат да ја завршат истата работа подобро од мене, никогаш нема да имаат можност поради немање пари воопшто да се обидат.

Алатките за големи податоци се разновидни. Ако имате време, речиси сигурно можете да напишете побрзо решение користејќи паметни техники за чистење, складирање и извлекување податоци. На крајот се сведува на анализа на трошоци и придобивки.

Што научив:

• не постои евтин начин да се анализираат 25 ТБ истовремено;
• Бидете внимателни со големината на вашите Паркет датотеки и нивната организација;
• Партициите во Spark мора да бидат избалансирани;
• Во принцип, никогаш не обидувајте се да направите 2,5 милиони партиции;
• Сортирањето е сè уште тешко, како и поставувањето на Spark;
• понекогаш специјалните податоци бараат посебни решенија;
• объединение в Spark работает быстро, но партиционирование всё ещё обходится дорого;
• не спиј кога те учат на основите, некој веројатно веќе ти го решил проблемот уште во 1980-тите;
• gnu parallel - ова е магична работа, секој треба да го користи;
• Спарк сака некомпресирани податоци и не сака комбинирање на партиции;
• Искра има премногу надземни трошоци при решавање на едноставни проблеми;
• Асоцијативните низи на AWK се многу ефикасни;
• можно обращаться к stdin и stdout од скрипта R, и затоа користете ја во нафтоводот;
• Благодарение на имплементацијата на паметната патека, S3 може да обработи многу датотеки;
• Главната причина за губење време е прерано оптимизирање на вашиот начин на складирање;
• не обидувајте се рачно да ги оптимизирате задачите, оставете го компјутерот да го направи тоа;
• API должен быть простым ради простоты и гибкости использования;
• Ако вашите податоци се добро подготвени, кеширањето ќе биде лесно!

Извор: www.habr.com