Databricks Delta を使用した大規模なゲノムパイプラインの簡素化

 <a href="https://www.databricks.com/resources/ebook/delta-lake-running-oreilly?itm_data=simplifyinggenomicspipelinesdelta-blog-oreillydlupandrunning">O'Reilly の新しい電子書籍の早期プレビュー</a>で、Delta Lake の使用を開始するために必要なステップバイステップのガイダンスをご確認ください。</p><hr><p><a class="dbce_cta" href="https://pages.databricks.com/rs/094-YMS-629/images/Simplifying%20Genomics%20Pipelines%20at%20Scale%20with%20Databricks.html" target="_blank">Databricks でこのノートブックを試す</a><br><em>このブログは、「大規模なゲノミクス分析」シリーズの最初のブログです。このシリーズでは、</em><a href="https://www.databricks.com/product/genomics"><em>Databricks 統合分析プラットフォーム for Genomics</em></a><em> を利用して顧客がどのように集団規模のゲノムデータを分析できるかをご紹介します。</em><a href="https://www.databricks.com/blog/2018/09/10/building-the-fastest-dnaseq-pipeline-at-scale.html"><em>ゲノミクスパイプライン</em></a><em>の出力を起点として、本シリーズでは Databricks を使用してサンプルの品質管理、共同ジェノタイピング、コホートの品質管理、高度な統計遺伝学分析を実行するためのチュートリアルを提供します。</em></p><hr><p>2003年の<a href="https://www.genome.gov/human-genome-project/What">ヒトゲノム計画</a>の完了以来、DNAシーケンシングのコストが最初のゲノムの30億ドル¹から今日では1,000ドル未満へと劇的に低下したことに後押しされ、データが爆発的に増加しました。</p><blockquote><p>[1] <a href="https://www.genome.gov/human-genome-project/What">ヒトゲノム計画</a>は、エネルギー省と国立衛生研究所が主導した30億ドルのプロジェクトで、1990年に開始され、2003年に完了しました。</p></blockquote><p><a href="https://www.databricks.com/wp-content/uploads/2019/03/costpergenome_2017.jpg"><img class="aligncenter wp-image-39985" style="width:700px;" src="https://www.databricks.com/wp-content/uploads/2019/03/costpergenome_2017.jpg" alt="ゲノムあたりのコスト" height="525"></a></p><p class="text-align-center">ソース：<a href="https://www.genome.gov/about-genomics/fact-sheets/DNA-Sequencing-Costs-Data">DNAシーケンシングのコスト：データ</a></p><p>その結果、ゲノミクス分野は、企業が人口規模でDNAシーケンシングを開始する段階まで成熟しました。しかし、DNAコードのシーケンシングは最初のステップにすぎず、その後、生データを分析に適したフォーマットに変換する必要があります。通常、これは一連の<a href="https://www.databricks.com/glossary/bioinformatics">バイオインフォマティクス</a>ツールをカスタムスクリプトでつなぎ合わせ、最終的にゲノムバリアントのコレクションが得られるまで、単一ノードで一度に1サンプルずつデータを処理することで行われます。 &nbsp;今日の<a href="https://www.databricks.com/glossary/bioinformatics">バイオインフォマティクス</a>科学者は、これらのパイプラインの構築と維持に時間の大部分を費やしています。ゲノムデータセットがペタバイト規模にまで拡大したことで、以下のような簡単な質問にさえ、タイムリーに答えることが困難になっています。</p><ul><li>今月、いくつのサンプルをシーケンスしましたか？</li><li>検出された固有バリアントの総数は？</li><li>さまざまなバリエーションのクラスで、いくつのバリアントが確認されましたか？</li></ul><p>さらにこの問題を複雑にしているのは、何千人もの個人からのデータが、アクセス可能でクエリ可能な状態を保ちながら、保存、追跡、バージョン管理ができないことです。その結果、研究者は解析を行う際にゲノムデータのサブセットを複製することが多く、全体的なストレージフットプリントとコストが急増します。 &nbsp;この問題を軽減するため、現在、研究者は通常6か月から2年間の「データフリーズ」という戦略を採用しています。この戦略では、新しいデータに関する作業を中断し、代わりに既存データの凍結されたコピーに集中します。より短い期間で段階的に解析を積み上げていくソリューションはなく、研究の進展が遅れる原因となっています。</p><p>産業規模でゲノムデータを処理できると同時に、科学者がデータを探索し、分析パイプラインを繰り返し実行し、新しい知見を引き出すための柔軟性を保持できる、堅牢なソフトウェアが強く求められています。</p><p><a href="https://www.databricks.com/wp-content/uploads/2019/06/genomics-reader-writer_image3.png"><img class="aligncenter wp-image-44884" style="width:698px;" src="https://www.databricks.com/wp-content/uploads/2019/06/genomics-reader-writer_image3.png" alt="Databricks を使用したエンドツーエンド ゲノミクス解析のためのアーキテクチャ" height="212"></a></p><p class="text-align-center">図1。Databricksを使用したエンドツーエンドのゲノミクス分析のアーキテクチャ</p><p><a href="https://www.youtube.com/watch?v=gg-lEJ4sBaA">Databricks Delta: リアルタイム ビッグ データ分析のための統合管理システム</a>により、Databricks プラットフォームは、今日の研究者が直面しているデータ ガバナンス、データ アクセス、データ分析の問題の解決に向けて大きなステップを踏み出しました。<a href="https://www.databricks.com/product/delta-lake-on-databricks">Databricks Delta Lake</a> を使用すると、すべてのゲノムデータを 1 か所に保存し、新しいデータが取り込まれるとリアルタイムで更新される分析を作成できます。ゲノミクス ファイル形式の読み取り、書き込み、処理を最適化する当社の <a href="https://www.databricks.com/product/genomics">ゲノミクス向け統合分析プラットフォーム</a> (UAP4G) と組み合わせることで、ゲノミクス パイプラインのワークフローにエンドツーエンドのソリューションを提供します。UAP4G アーキテクチャは柔軟性を提供し、顧客が独自のパイプラインをプラグインして、独自の三次アナリティクスを開発できるようにします。例として、品質管理メトリクスと視覚化を表示する次のダッシュボードを紹介します。これらは自動的に計算および表示され、特定の要件に合わせてカスタマイズできます。</p><p>https://www.youtube.com/watch?v=73fMhDKXykU</p><p>このブログの残りの部分では、サンプル処理の完了に応じてリアルタイムで更新される、上記の品質管理ダッシュボードを構築したステップを順を追って説明します。ゲノムデータ処理に Delta ベースのパイプラインを使用することで、お客様はリアルタイムでサンプルごとの可視性を確保できる方法でパイプラインを運用できるようになります。Databricks ノートブック（およびGitHubやMLflowなどのインテグレーション）を使用することで、結果の再現性を確保できる方法で分析を追跡し、バージョン管理することができます。バイオインフォマティシャンは、パイプラインの維持にかける時間を減らし、新たな発見により多くの時間を費やすことができるようになります。私たちはUAP4Gを、アドホックな分析を産業規模の本番ゲノミクスへと変革させるエンジンと見なしています。これにより、遺伝学と疾患のlinkについて、より深い知見が可能になります。</p><h2>サンプルデータの読み込み</h2><p>まず、サンプルの小規模なコホートからバリアントデータを読み込むことから始めましょう。次のステートメントは、特定の sampleId のデータを読み込み、Databricks Delta 形式で (delta_stream_output フォルダーに) 保存します。</p><pre style="font-size:10pt;">spark.read.\ format("parquet").\ load("dbfs:/annotations_etl_parquet/sampleId=" + "SRS000030_SRR709972").\ write.\ format("delta"). \ save(delta_stream_outpath) </pre><blockquote><p>注: annotations_etl_parquet フォルダーには、<a href="https://www.internationalgenome.org/">1000ゲノムデータセット</a>から生成され、Parquet 形式で格納されたアノテーションが含まれています。 &nbsp;&nbsp;これらのアノテーションのETLと処理は、<a href="https://www.databricks.com/product/genomics">Databricks’ Unified Analytics Platform for Genomics</a> を使用して実行されました。</p></blockquote><h2>Databricks Delta テーブルのストリーミングを起動</h2><p>次のステートメントでは、Databricks Delta 形式を使用してデータのストリーム (readStream 経由) を読み取る exomes Apache Spark DataFrame を作成しています。 &nbsp;これは、継続的に実行される動的な DataFrame です。つまり、データが delta_stream_output フォルダーに書き込まれると、exomes DataFrame は新しいデータを読み込みます。 &nbsp;&nbsp;エクソーム DataFrame を表示するには、DataFrame クエリをランして、sampleId でグループ化されたバリアントの数を調べます。</p><pre style="font-size:10pt;"># Read the stream of data exomes = spark.readStream.format("delta").load(delta_stream_outpath) # Display the data via DataFrame query display(exomes.groupBy("sampleId").count().withColumnRenamed("count", "variants")) </pre><p><code>display</code> ステートメントを実行すると、Databricks ノートブックは、ストリーミング ジョブを監視するためのストリーミング ダッシュボードを提供します。ストリーミング ジョブのすぐ下に、display ステートメントの結果 (つまり、sample_id ごとのバリアント数) が表示されます。</p><p><a href="https://www.databricks.com/wp-content/uploads/2019/03/single_sample_display_variant_count_by_sampleid.png"><img class="aligncenter wp-image-39992" style="width:700px;" src="https://www.databricks.com/wp-content/uploads/2019/03/single_sample_display_variant_count_by_sampleid.png" alt height="282"></a></p><p><code>`exomes`</code> DataFrame に基づいて他の DataFrame クエリーを実行し、最初の質問セットに引き続き答えていきましょう。</p><h2>一塩基バリアント数</h2><p>例を続けると、次のグラフに表示されるように、一塩基バリアント (SNV) の数を迅速に計算できます。</p><pre style="font-size:10pt;">%sql select referenceAllele, alternateAllele, count(1) as GroupCount from snvs group by referenceAllele, alternateAllele order by GroupCount desc </pre><p><a href="https://www.databricks.com/wp-content/uploads/2019/03/single_nucleotide_variant_count_sorted.png"><img class="aligncenter wp-image-39994" style="width:699px;" src="https://www.databricks.com/wp-content/uploads/2019/03/single_nucleotide_variant_count_sorted.png" alt height="356"></a></p><blockquote><p>注：<code>display</code> コマンドは Databricks ワークスペースの一部であり、Databricks の可視化機能を使用して DataFrame を表示できます（すなわち、コーディングは不要です）。</p></blockquote><h2>バリアント数</h2><p>バリアントに機能的効果をアノテーション付けしたので、見られるバリアント効果の広がりを確認することで分析を続けることができます。検出されたバリアントの大部分は、タンパク質をコードする領域に隣接しており、これらは非コードバリアントとして知られています。</p><pre style="font-size:10pt;">display(exomes.groupBy("mutationType").count()) </pre><p><a href="https://www.databricks.com/wp-content/uploads/2019/03/mutation_type_donut.png"><img class="aligncenter wp-image-39996" style="width:650px;" src="https://www.databricks.com/wp-content/uploads/2019/03/mutation_type_donut.png" alt height="534"></a></p>

アミノ酸置換ヒートマップ

引き続き exomes DataFrame を使用して、次のコード スニペットでアミノ酸置換数を計算しましょう。 以前の DataFrames と同様に、別の動的 DataFrame (aa_counts) を作成します。これにより、exomes DataFrame で新しいデータが処理されるのに応じて、アミノ酸置換数にもその結果が反映されます。  また、データをメモリに書き込んでいます (つまり、.format(“memory”))60 秒ごとにバッチを処理し (つまり、trigger(processingTime=’60 seconds’)) により、下流の Pandas ヒートマップ コードがヒートマップを処理して可視化できます。

次のコードスニペットは、先行するamino_acid_substitutions Spark テーブルを読み取り、最大カウントを決定し、Spark テーブルから新しい Pandas ピボットテーブルを作成して、ヒートマップをプロットします。

継続的パイプラインへの移行

ここまで、先行するコードスニペットと可視化は、単一の sampleId に対する単一のランを表しています。  しかし、Structured StreamingとDatabricks Deltaを使用しているため、このコードを（変更なしで）使用して、サンプルがパイプラインを通過するたびに品質管理統計を継続的に計算する本番運用データパイプラインを構築できます。これを実証するために、データセット全体をロードする次のコードスニペットを実行できます。

以前のコードスニペットで説明したように、exomes DataFrame のソースは delta_stream_output フォルダに読み込まれたファイルです。  最初は、単一の sampleId (すなわち sampleId = “SRS000030_SRR709972”) のファイルセットを読み込みました。 前のコードスニペットは、生成されたすべての Parquet サンプル（すなわち、Parquet）を取得し、それらのファイルを sampleId ごとに同じ delta_stream_output フォルダに増分的に読み込みます。   以下のアニメーションGIFは、前のコードスニペットの省略された出力を示しています。

https://www.youtube.com/watch?v=JPngSC5Md-Q

ゲノミクス パイプラインの可視化

ノートブックの先頭までスクロールして戻ると、exomes DataFrame が新しい sampleIds を自動的に読み込んでいることに気付くでしょう。 ゲノミクスパイプラインの構造化ストリーミングコンポーネントは継続的に実行されるため、新しいファイルが delta_stream_outputpath フォルダに読み込まれるとすぐにデータを処理します。  Databricks Delta フォーマットを使用することで、exomes DataFrame にストリーミングされるデータのトランザクションの整合性を確保できます。

https://www.youtube.com/watch?v=Q7KdPsc5mbY

exomes DataFrame の初期作成時とは対照的に、構造化ストリーミングのモニタリング ダッシュボードが現在データをロードしている様子に注目してください（すなわち、「input vs. processing rate」の変動、「バッチ期間」の変動、および「aggregations state」における個別キーの増加）。  exomes DataFrame が処理されるにつれて、sampleIds (およびバリアント数) の新しい行に注目してください。この同じアクションは、関連するグループ by mutation type クエリーでも見られます。

https://www.youtube.com/watch?v=sT179SCknGM

Databricks Delta により、当社のゲノミクス パイプラインの各ステップで、あらゆる新規データのトランザクション上の一貫性が保たれます。  これが重要なのは、パイプラインの一貫性 (データの一貫性を維持し、すなわち全データが「正しい」ことを保証)、信頼性 (トランザクションは完全に成功するか、さもなければ完全に失敗する)、リアルタイム更新への対応 (多数のトランザクションを並行して処理でき、いかなる停止や障害もデータに影響を与えない能力) が保証されるためです。したがって、(多数の追加 ETL ステップがあった) 下流のアミノ酸置換マップのデータでさえ、シームレスに更新されます。

ゲノミクスパイプラインの最終ステップとして、DBFS内のDatabricks Delta parquetファイルを確認し、固有の変異（経時的な固有の変異の増加）もモニタリングしています。

まとめ

Databricks統合分析プラットフォームの基盤、特にDatabricks Deltaを利用することで、バイオインフォマティシャンや研究者はDatabricks Unified Analytics Platform for ゲノミクスを使用して、トランザクションの一貫性を保った分散アナリティクスを適用できます。これらの抽象化により、データ実務者はゲノミクス パイプラインを簡素化できます。  ここでは、手動での介入なしに、新しいサンプルが処理されるたびにデータを継続的に処理する、ゲノムサンプルの品質管理パイプラインを作成しました。  ETLを実行している場合でも、高度なアナリティクスを実行している場合でも、データはゲノミクス パイプラインを通じて迅速かつ中断なく流れます。今すぐ「Simplifying ゲノミクス Pipelines at Scale with Databricks Delta ノートブック」をダウンロードしてお試しください。

大規模ゲノミクス分析の開始：

• Unified Analytics for ゲノミクス向け統合分析のソリューション ガイドをご覧ください
• Databricks Deltaによる大規模ゲノミクスパイプラインの簡素化ノートブックをダウンロード
• Databricks Unified Analytics for Genomics のフリートライアルにサインアップ

謝辞

Yongsheng HuangとMichael Ortegaの貢献に感謝します。