この記事の読み方: 非常に長くて混沌とした文章になってしまい申し訳ありません。時間を節約するために、各章は「学んだこと」の導入部から始まり、その章の要点を XNUMX ～ XNUMX 文で要約しています。

「解決策を教えてください！」 私の原点を知りたいだけなら、「もっと創意工夫する」の章まで飛ばしてください。ただし、失敗について読んだほうが興味深く、役に立つと思います。

私は最近、大量の生の DNA 配列 (技術的には SNP チップ) を処理するプロセスをセットアップする任務を負っていました。必要なのは、後続のモデリングやその他のタスクのために、特定の遺伝子位置 (SNP と呼ばれる) に関するデータを迅速に取得することでした。 R と AWK を使用すると、自然な方法でデータを整理して整理することができ、クエリ処理が大幅に高速化されました。これは私にとって簡単ではなく、何度も繰り返す必要がありました。この記事は、私の間違いのいくつかを回避し、私が最終的にどうなったかを示すのに役立ちます。

まず、入門的な説明をします。

データ

私たちの大学の遺伝情報処理センターからは、25 TB TSV の形式でデータが提供されました。私はそれらを 5 つのパッケージに分割して受け取り、Gzip で圧縮しました。各パッケージには約 240 の 2,5 GB のファイルが含まれていました。各行には、60 人の個人からの 30 つの SNP のデータが含まれていました。合計で、約 XNUMX 万の SNP と約 XNUMX 万人に関するデータが送信されました。ファイルには、SNP 情報に加えて、読み取り強度、さまざまな対立遺伝子の頻度など、さまざまな特性を反映する数値が記載された多数の列が含まれていました。一意の値を持つ列は合計で約 XNUMX 列ありました。

目標

他のデータ管理プロジェクトと同様に、最も重要なことは、データがどのように使用されるかを決定することでした。この場合 主に SNP に基づいて SNP のモデルとワークフローを選択します。。つまり、一度に 2,5 つの SNP に関するデータのみが必要になります。私は、XNUMX 万の SNP の XNUMX つに関連付けられたすべてのレコードを、できるだけ簡単、迅速、そして安価に取得する方法を学ばなければなりませんでした。

これをしない方法

適切な決まり文句を引用すると、次のようになります。

私は何千回も失敗したわけではありません。クエリに適した形式で大量のデータを解析することを回避する千の方法を発見しただけです。

最初に試す

何を学んだのか: 一度に 25 TB を解析する安価な方法はありません。

ヴァンダービルト大学で「ビッグデータ処理の高度な手法」コースを受講したので、私はそのコツがバッグの中にあると確信していました。 Hive サーバーをセットアップしてすべてのデータを実行し、結果をレポートするには、おそらく 3 ～ XNUMX 時間かかります。データは AWS SXNUMX に保存されているため、サービスを使用しましたアテナこれにより、Hive SQL クエリを S3 データに適用できるようになります。 Hive クラスターをセットアップ/構築する必要はなく、探しているデータに対してのみ料金を支払います。

Athena にデータとその形式を示した後、次のようなクエリを使用していくつかのテストを実行しました。

select * from intensityData limit 10;

そして、すぐによく構造化された結果を受け取りました。準備ができて。

データを仕事に使おうとするまでは...

モデルをテストするためにすべての SNP 情報を引き出すように求められました。クエリを実行しました:


select * from intensityData 
where snp = 'rs123456';

...そして待ち始めました。 4 分後、5 TB を超えるデータが要求された後、結果を受け取りました。 Athena は、見つかったデータの量に応じて 20 テラバイトあたり 38 ドルを請求します。つまり、この 50 回のリクエストには XNUMX ドルのコストと XNUMX 分の待機時間がかかりました。すべてのデータに対してモデルを実行するには、XNUMX 年間待機し、XNUMX 万ドルを支払わなければなりませんでしたが、これは明らかに私たちには適していませんでした。

寄木細工を使用する必要がありました...

何を学んだのか: Parquet ファイルのサイズとその構成には注意してください。

私は最初に、すべての TSV を次の形式に変換することで状況を修正しようとしました。寄木細工のファイル。これらは、その中の情報が列形式で保存されるため、大規模なデータセットを扱うのに便利です。行に各列の要素が含まれるテキストファイルとは対照的に、各列は独自のメモリ/ディスクセグメントに存在します。何かを見つける必要がある場合は、必要な列を読んでください。さらに、各ファイルは列に値の範囲を保存するため、探している値が列の範囲にない場合でも、Spark はファイル全体のスキャンに時間を無駄にすることはありません。

簡単なタスクを実行しました AWSグルー TSV を Parquet に変換し、新しいファイルを Athena にドロップしました。約5時間かかりました。しかし、リクエストを実行してみると、完了までにかかる時間はほぼ同じで、費用は少し少なくなりました。実際のところ、Spark はタスクを最適化しようとして、XNUMX つの TSV チャンクを解凍し、それを独自の Parquet チャンクに配置しただけです。また、各チャンクは多くの人の記録全体を含めるのに十分な大きさで、各ファイルにはすべての SNP が含まれていたため、Spark は必要な情報を抽出するためにすべてのファイルを開く必要がありました。

興味深いことに、Parquet のデフォルト (および推奨) 圧縮タイプである snappy は分割可能ではありません。そのため、各実行者は、3,5 GB の完全なデータセットを解凍してダウンロードするという作業に追われていました。

問題を理解しましょう

何を学んだのか: 特にデータが分散されている場合、並べ替えは困難です。

今、私は問題の本質を理解したように思えました。データを人ごとではなく、SNP 列ごとに並べ替えるだけで済みました。次に、いくつかの SNP が別のデータチャンクに保存され、Parquet の「スマート」機能「値が範囲内にある場合にのみオープン」がその栄光を発揮します。残念ながら、クラスター内に散在する数十億行を並べ替えるのは困難な作業であることが判明しました。

大学でアルゴリズムの授業を受けている私: 「うーん、これらすべての並べ替えアルゴリズムの計算の複雑さなど誰も気にしません。」

20TB の列でソートしようとしている＃スパーク表: 「なぜこんなに時間がかかるのですか?」 #データサイエンス闘争。

— ニック・ストレイヤー (@NicholasStrayer) 2019 年 3 月 11 日

AWS は、「集中力のない学生だから」という理由で返金を行うことを絶対に望んでいません。 Amazon Glue で並べ替えを実行した後、2 日間実行されてクラッシュしました。

パーティション分割についてはどうですか?

何を学んだのか: Spark のパーティションはバランスをとる必要があります。

そこで私は、データを染色体で分割するというアイデアを思いつきました。それらは 23 個あります (ミトコンドリア DNA とマッピングされていない領域を考慮するとさらにいくつかあります)。
これにより、データをより小さなチャンクに分割できるようになります。 Glue スクリプトの Spark エクスポート関数に XNUMX 行だけ追加すると、 partition_by = "chr"、その後、データをバケットに分割する必要があります。

ゲノムは染色体と呼ばれる多数の断片から構成されています。

残念ながら、うまくいきませんでした。染色体のサイズが異なるということは、情報量が異なることを意味します。これは、一部のノードが早く終了してアイドル状態になったため、Spark がワーカーに送信したタスクのバランスがとれず、完了に時間がかかることを意味します。ただし、タスクは完了しました。しかし、10 つの SNP を要求すると、不均衡が再び問題を引き起こしました。より大きな染色体 (つまり、データを取得したい場所) 上の SNP を処理するコストは、約 XNUMX 分の XNUMX しか減少しません。たくさんありますが、十分ではありません。

さらに細かく分割するとどうなるでしょうか？

何を学んだのか: 2,5 万パーティションを決して実行しないでください。

私は思い切って各 SNP を分割することにしました。これにより、パーティションのサイズが確実に同じになります。 それは悪い考えでした。接着剤を使って無邪気なラインを追加しました partition_by = 'snp'。タスクが開始され、実行が開始されました。 3 日後確認したところ、まだ S3 に何も書き込まれていなかったので、タスクを強制終了しました。 Glue が SXNUMX の隠し場所に中間ファイルを書き込んでいたようです。おそらく数百万もの大量のファイルが書き込まれていました。結果として、私の間違いは千ドル以上の損害をもたらし、私の指導者を喜ばせることはできませんでした。

パーティショニング + ソート

何を学んだのか: Spark のチューニングと同様に、並べ替えは依然として困難です。

前回のパーティション化の試みでは、染色体をパーティション化し、各パーティションを並べ替えました。理論的には、必要な SNP データは指定された範囲内のいくつかの Parquet チャンク内にある必要があるため、これにより各クエリが高速化されます。残念ながら、分割されたデータであっても並べ替えるのは困難な作業であることが判明しました。その結果、カスタムクラスターの EMR に切り替え、5.4 つの強力なインスタンス (CXNUMXxl) と Sparklyr を使用して、より柔軟なワークフローを作成しました。

# Sparklyr snippet to partition by chr and sort w/in partition
# Join the raw data with the snp bins
raw_data
  group_by(chr) %>%
  arrange(Position) %>% 
  Spark_write_Parquet(
    path = DUMP_LOC,
    mode = 'overwrite',
    partition_by = c('chr')
  )

...しかし、タスクはまだ完了していませんでした。各クエリ実行プログラムのメモリ割り当てを増やしたり、大量のメモリを持つノードを使用したり、ブロードキャスト変数 (ブロードキャスト変数) を使用したり、さまざまな方法で構成しましたが、毎回これらは中途半端であることが判明し、徐々に実行プログラムが起動し始めました。すべてが止まるまで失敗すること。

アップデート：それで始まります。 pic.twitter.com/agY4GU2ru5

— ニック・ストレイヤー (@NicholasStrayer) 2019 年 5 月 15 日

より創造的になっていきます

何を学んだのか: 特殊なデータには特殊なソリューションが必要な場合があります。

各 SNP には位置の値があります。これは、染色体上の塩基の数に対応する数値です。これはデータを整理するための自然で優れた方法です。最初は各染色体の領域ごとに分割したいと考えていました。たとえば、位置 1 ～ 2000、2001 ～ 4000 などです。しかし問題は、SNP が染色体全体に均一に分布していないため、グループのサイズが大きく異なることです。

その結果、役職をカテゴリー（ランク）に分類することにたどり着きました。すでにダウンロードしたデータを使用して、固有の SNP、その位置、および染色体のリストを取得するリクエストを実行しました。次に、各染色体内のデータを並べ替え、SNP を所定のサイズのグループ (ビン) に収集しました。それぞれ 1000 の SNP があるとします。これにより、SNP と染色体ごとのグループの関係がわかりました。

最終的に 75 個の SNP のグループ (bin) を作成しました。その理由は以下で説明します。

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の追加

何を学んだのか：基本を教えてもらっている間は寝ないでください。きっと誰かが 1980 年代にすでにあなたの問題を解決していました。

この時点まで、Spark でのすべての失敗の原因はクラスター内のデータのごちゃ混ぜでした。おそらく、前処理によって状況が改善される可能性があります。私は生のテキストデータを染色体の列に分割してみることにしたので、Spark に「事前に分割された」データを提供したいと考えました。

StackOverflowで列の値で分割する方法を検索したところ、とても素晴らしい答えです。 AWK を使用すると、結果を送信するのではなく、スクリプトにテキストファイルを記述することで、列の値ごとにテキストファイルを分割できます。 stdout.

Bash スクリプトを書いて試してみました。パッケージ化された TSV の XNUMX つをダウンロードし、次のコマンドを使用して解凍します。 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 つのコアと約 XNUMX MB のメモリのみを使用していることが判明しました。多額の損失を出さずに問題を解決するには、作業を並列化する方法を見つける必要がありました。幸いなことに、本当に素晴らしい本の中でコマンドラインでのデータサイエンス Jeron Janssens による並列化に関する章を見つけました。そこから私が学んだことは、 gnu parallel、Unix でマルチスレッドを実装するための非常に柔軟な方法です。

新しいプロセスを使用してパーティショニングを開始したときは、すべて問題ありませんでしたが、まだボトルネックがありました。S3 オブジェクトのディスクへのダウンロードはそれほど高速ではなく、完全に並列化されていませんでした。これを修正するために、次のようにしました。

S3 ダウンロードステージをパイプラインに直接実装し、ディスク上の中間ストレージを完全に排除できることがわかりました。これは、生データをディスクに書き込む必要がなく、AWS 上のさらに小さい、したがって安価なストレージを使用できることを意味します。
チーム aws configure set default.s3.max_concurrent_requests 50 AWS CLI が使用するスレッドの数が大幅に増加しました (デフォルトでは 10 です)。
名前に文字 n が含まれる、ネットワーク速度に最適化された EC2 インスタンスに切り替えました。 n インスタンスを使用する場合の処理能力の損失は、読み込み速度の向上によって十分に補われることがわかりました。ほとんどのタスクでは c5n.4xl を使用しました。
かわった 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 のおかげで、19 ギガの CSV をダウンロードするのと同じくらい速く解凍して分割できます。これを実行するための火花さえ得られませんでした。 #データサイエンス #Linux pic.twitter.com/Nqyba2zqEk

— ニック・ストレイヤー (@NicholasStrayer) 2019 年 5 月 17 日

このツイートには「TSV」について言及する必要がありました。ああ、ああ。

新しく解析されたデータの使用

何を学んだのか: Spark は非圧縮データを好み、パーティションの結合を好みません。

これで、データがアンパック (共有: 共有) かつ半順序付けされた形式で S3 に保存され、再び Spark に戻ることができました。驚きが私を待っていました。私は再び望んでいたものを達成できませんでした。データがどのように分割されているかを Spark に正確に伝えるのは非常に困難でした。そして、これを実行したときでも、パーティションが多すぎることが判明しました（95）。 coalesce その数を妥当な制限まで減らしたので、パーティション分割が破壊されました。これは修正できると確信していますが、数日間検索しても解決策が見つかりませんでした。最終的には Spark ですべてのタスクを完了しましたが、時間がかかり、分割された Parquet ファイルはそれほど小さくありませんでした (~200 KB)。ただし、データは必要な場所にありました。

小さすぎて凸凹してて素敵！

ローカル Spark クエリのテスト

何を学んだのか: 単純な問題を解決する場合、Spark にはオーバーヘッドが多すぎます。

データを賢い形式でダウンロードすることで、速度をテストすることができました。ローカル Spark サーバーを実行するように R スクリプトを設定し、指定された Parquet グループストレージ (bin) から Spark データフレームをロードします。すべてのデータをロードしようとしましたが、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 秒かかりました。はるかに優れていますが、何かを大量にテストするにはあまり適していません。さらに、データフレームをメモリにキャッシュしようとすると、重みが 50 未満のデータセットに 15 GB 以上のメモリを割り当てた場合でも、Spark が常にクラッシュしたため、キャッシュを使用しても速度を上げることができませんでした。

AWKに戻る

何を学んだのか: AWK の連想配列は非常に効率的です。

もっと高速に到達できることに気づきました。それを素晴らしい形で思い出しました Bruce Barnett による AWK チュートリアル「」という素晴らしい機能について読みました。連想配列」基本的に、これらはキーと値のペアですが、何らかの理由で AWK では異なる呼び方をしていたので、私はそれらについてあまり考えていませんでした。ローマン・チェプリャカ「連想配列」という用語は「キーと値のペア」という用語よりもはるかに古いことを思い出しました。たとえあなたが Google Ngram で Key-Value を検索するこの用語はそこにはありませんが、連想配列は見つかります。さらに、「キーと値のペア」はデータベースに関連付けられることが多いため、ハッシュマップと比較することは非常に合理的です。これらの連想配列を使用すると、Spark を使用せずに SNP をビンテーブルおよび生データに関連付けることができることに気付きました。

これを行うには、AWK スクリプトで次のブロックを使用しました。 BEGIN。これは、データの最初の行がスクリプトの本体に渡される前に実行されるコードです。

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

チーム while(getline...) CSV グループ (bin) からすべての行をロードし、最初の列 (SNP 名) を連想配列のキーとして設定します bin 値として XNUMX 番目の値 (グループ) を指定します。それからブロックで { }これはメインファイルのすべての行で実行され、各行は出力ファイルに送信され、グループ (bin) に応じて一意の名前が付けられます。 ..._bin_"bin[$1]"_....

変数 batch_num и chunk_id パイプラインによって提供されるデータと一致し、競合状態が回避され、各実行スレッドが実行されます。 parallel、独自の一意のファイルに書き込みます。

AWK を使用した前回の実験で残った染色体上のフォルダーにすべての生データを分散したため、一度に 3 つの染色体を処理し、より深く分割されたデータを SXNUMX に送信する別の Bash スクリプトを作成できるようになりました。

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/*

スクリプトには XNUMX つのセクションがあります parallel.

最初のセクションでは、目的の染色体に関する情報を含むすべてのファイルからデータが読み取られ、次にこのデータがスレッド間で分散され、ファイルが適切なグループ (bin) に分散されます。複数のスレッドが同じファイルに書き込むときの競合状態を回避するために、AWK はファイル名を渡してデータを別の場所に書き込みます。 chr_10_bin_52_batch_2_aa.csv。その結果、ディスク上に小さなファイルが多数作成されます (このために、テラバイトの EBS ボリュームを使用しました)。

第二セクションからのコンベヤー parallel グループ (bin) を調べて、個々のファイルを共通の CSV c に結合します。 catそしてエクスポートのために送信します。

Rで放送？

何を学んだのか：連絡できる stdin и stdout R スクリプトから取得するため、パイプラインで使用します。

Bash スクリプトに次の行があることに気付いたかもしれません。 ...cat chunked/*_bin_{}_*.csv | ./upload_as_rds.R...。すべての連結されたグループファイル (bin) が以下の 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 の部分は高度に最適化されています。 5 つの中型染色体でテストした後、C4n.XNUMXxl インスタンスでのジョブは約 XNUMX 時間で完了しました。

S3 の制限事項

何を学んだのか: スマートパスの実装のおかげで、S3 は多くのファイルを処理できます。

S3 に転送される多数のファイルを S3 が処理できるかどうかが心配でした。意味のあるファイル名にすることはできますが、SXNUMX はどのようにファイル名を探すのでしょうか?

S3 のフォルダーは単なる見せ物であり、実際にはシステムはこのシンボルには興味がありません。 /. S3 FAQ ページより。

S3 は、特定のファイルへのパスを、一種のハッシュテーブルまたはドキュメントベースのデータベース内の単純なキーとして表しているようです。バケットはテーブルとして考えることができ、ファイルはそのテーブル内のレコードとして考えることができます。

Amazon で利益を上げるにはスピードと効率が重要であるため、このファイルパスとしてのキーシステムが異常に最適化されていることは驚くべきことではありません。大量の取得リクエストを作成する必要がなく、リクエストが迅速に実行されるように、バランスを見つけようとしました。約20万個のbinファイルを作成するのが最適であることが判明しました。最適化を続ければ、速度の向上が達成できると思います (たとえば、データ専用の特別なバケットを作成して、ルックアップテーブルのサイズを削減するなど)。しかし、さらなる実験を行うための時間もお金もありませんでした。

相互互換性についてはどうですか?

学んだこと: 時間を無駄にする最大の原因は、保管方法を時期尚早に最適化することです。

この時点で、「なぜ独自のファイル形式を使用するのか?」と自問することが非常に重要です。その理由は、読み込み速度 (gzip 圧縮された CSV ファイルの読み込みに 7 倍の時間がかかりました) とワークフローとの互換性にあります。 R が Spark をロードせずに Parquet (または Arrow) ファイルを簡単にロードできるかどうか、再考するかもしれません。私たちの研究室では全員が R を使用しており、データを別の形式に変換する必要がある場合でも、元のテキストデータがまだ残っているので、パイプラインを再度実行するだけで済みます。

仕事の分割

何を学んだのか: ジョブを手動で最適化しようとせず、コンピューターに任せてください。

XNUMX つの染色体でワークフローをデバッグしましたが、今度は他のすべてのデータを処理する必要があります。
変換のために複数の EC2 インスタンスを起動したいと考えていましたが、同時に、(Spark が不均衡なパーティションに悩まされたのと同じように) 異なる処理ジョブ間で非常に不均衡な負荷が発生するのではないかと心配していました。さらに、AWS アカウントにはデフォルトでインスタンス数が 10 個に制限されているため、染色体ごとに XNUMX つのインスタンスを生成することに興味がありませんでした。

そこで、処理ジョブを最適化するスクリプトを 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 を使用して XNUMX 回のシャッフルを実行し、最良のものを選択しました。

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 Parquet ファイルに変換する場合、これは同僚ではなく私にとって問題になるはずです。このために、内部 R パッケージを作成することにしました。

関数を中心に編成されたいくつかのデータアクセス関数のみを含む非常に単純なパッケージを構築して文書化する get_snp。同僚向けのウェブサイトも作りましたパッケージダウン、例やドキュメントを簡単に見ることができます。

スマートキャッシング

何を学んだのか: データが十分に準備されていれば、キャッシュは簡単です。

主要なワークフローの XNUMX つが同じ分析モデルを SNP パッケージに適用したため、ビニングを有利に使用することにしました。 SNP 経由でデータを送信する場合、グループ (bin) からのすべての情報が返されるオブジェクトに添付されます。つまり、古いクエリは (理論上は) 新しいクエリの処理を高速化できます。

# Part of get_snp()
...
  # Test if our current snp data has the desired snp.
  already_have_snp <- desired_snp %in% prev_snp_results$snps_in_bin

  if(!already_have_snp){
    # Grab info on the bin of the desired snp
    snp_results <- get_snp_bin(desired_snp)

    # Download the snp's bin data
    snp_results$bin_data <- aws.s3::s3readRDS(object = snp_results$data_loc)
  } else {
    # The previous snp data contained the right bin so just use it
    snp_results <- prev_snp_results
  }
...

パッケージを構築するときに、さまざまな方法を使用した場合の速度を比較するために多くのベンチマークを実行しました。予期せぬ結果が生じる場合があるため、これを無視しないことをお勧めします。例えば、 dplyr::filter インデックスベースのフィルタリングを使用して行をキャプチャするよりもはるかに高速であり、フィルタリングされたデータフレームから単一の列を取得するのは、インデックス構文を使用するよりもはるかに高速でした。

オブジェクトに注意してください prev_snp_results キーが含まれています snps_in_bin。これはグループ (ビン) 内のすべての一意の SNP の配列であり、以前のクエリからのデータが既に存在するかどうかをすばやく確認できます。また、次のコードを使用すると、グループ (ビン) 内のすべての 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 
}

結果

今では、以前はアクセスできなかったモデルやシナリオを実行できるようになりました (そして本格的に実行し始めています)。一番良いのは、研究室の同僚が複雑なことを考える必要がないことです。機能する機能があるだけです。

パッケージでは詳細は省いていますが、明日私が突然いなくなった場合でも理解できるように、データ形式をシンプルにするよう努めました...

速度が著しく向上しました。私たちは通常、機能的に重要なゲノム断片をスキャンします。以前はこれを実行できませんでした (コストが高すぎることが判明しました)。しかし、現在はグループ (ビン) 構造とキャッシュのおかげで、0,1 つの SNP のリクエストにかかる時間は平均 3 秒未満で、データ使用量も大幅に削減されています。 SXNUMX のコストはピーナッツほど低いです。

最近、研究室で 25 TB 以上の生のジェノタイピングデータを処理する必要がありました。私が使い始めたとき、spark の使用には 8 分かかり、SNP のクエリに 20 ドルの費用がかかりました。 AWK+使用後 #rstats 処理にかかる時間は 10 分の 0.00001 秒未満で、コストは XNUMX ドルです。個人的な＃ビッグデータ勝つ。 pic.twitter.com/ANOXVGrmkk

— ニック・ストレイヤー (@NicholasStrayer) 2019 年 5 月 30 日

まとめ

この記事はまったくガイドではありません。解決策は個別のものであり、ほぼ確実に最適ではないことが判明しました。というより旅行記です。他の人たちには、そのような決定は頭の中で完全に形成されたものではなく、試行錯誤の結果であることを理解してもらいたいと思います。また、データサイエンティストを探している場合は、これらのツールを効果的に使用するには経験が必要であり、経験には費用がかかることに留意してください。私にはお金を払う余裕があったので幸せですが、私よりも同じ仕事ができる他の多くの人は、お金がないために挑戦する機会すらないでしょう。

ビッグデータツールは多用途です。時間があれば、スマートなデータクリーニング、保存、抽出手法を使用して、より高速なソリューションを作成することがほぼ確実に可能です。最終的には費用対効果の分析に行き着きます。

私が学んだこと:

一度に 25 TB を解析する安価な方法はありません。
Parquet ファイルのサイズとその構成には注意してください。
Spark のパーティションはバランスをとる必要があります。
一般に、2,5 万個のパーティションを作成しようとしないでください。
Spark のセットアップと同様に、並べ替えは依然として困難です。
特殊なデータには特殊なソリューションが必要な場合があります。
Spark の集約は高速ですが、パーティショニングには依然としてコストがかかります。
基本を教えるときは眠らないでください。おそらく 1980 年代に誰かがすでにあなたの問題を解決していました。
gnu parallel - これは魔法のようなものです。誰もがそれを使うべきです。
Spark は非圧縮データを好み、パーティションの結合を好みません。
単純な問題を解決する場合、Spark のオーバーヘッドは多すぎます。
AWK の連想配列は非常に効率的です。
連絡できる stdin и stdout R スクリプトから取得するため、パイプラインで使用します。
スマートパスの実装のおかげで、S3 は多くのファイルを処理できます。
時間を無駄にする主な理由は、保管方法を時期尚早に最適化することです。
タスクを手動で最適化しようとせず、コンピュータに任せてください。
使いやすさと柔軟性を高めるために、API はシンプルである必要があります。
データが十分に準備されていれば、キャッシュは簡単です。

出所： habr.com

AWK と R を使用して 25TB を解析する

データ

目標

これをしない方法

最初に試す

寄木細工を使用する必要がありました...

問題を理解しましょう

パーティション分割についてはどうですか?

さらに細かく分割するとどうなるでしょうか？

パーティショニング + ソート

より創造的になっていきます

まずはSparkで試してみる

AWKの追加

コアを埋める

新しく解析されたデータの使用

ローカル Spark クエリのテスト

AWKに戻る

Rで放送？

S3 の制限事項

相互互換性についてはどうですか?

仕事の分割

梱包しましょう！

スマートキャッシング

結果

まとめ

私が学んだこと:

コメントを追加します返信をキャンセル

AWK と R を使用して 25TB を解析する

データ

目標

これをしない方法

最初に試す

寄木細工を使用する必要がありました...

問題を理解しましょう

パーティション分割についてはどうですか?

さらに細かく分割するとどうなるでしょうか？

パーティショニング + ソート

より創造的になっていきます

まずはSparkで試してみる

AWKの追加

コアを埋める

新しく解析されたデータの使用

ローカル Spark クエリのテスト

AWKに戻る

Rで放送？

S3 の制限事項

相互互換性についてはどうですか?

仕事の分割

梱包しましょう！

スマートキャッシング

結果

まとめ

私が学んだこと:

コメントを追加します 返信をキャンセル

コメントを追加します返信をキャンセル