3. Spark 技术介绍 Spark 是一种与 相似的集群计算环境， 但 Spark 启用了内存分布数据集， 在某些工作负载方面表现得更加优越，除了能够提供交互式查询外，Spark 还可以优化迭代工作负载。
Spark 的主要特点包括：
- 速度快 Spark 具有先进的 DAG 执行引擎，支持迭代数据流和内存内计算，应用程序执行速度是 Hadoop 在内存内 MapReduce 的 100 倍，或在磁盘上的 10 倍；
 - 易于使用 用 Java、Scala、Python 和 R 快速编写应用， Spark 提供超过 80 种高级操作，构建并行应用非常方便，而且可以与 Scala 、Python 和 R 进行交互；
- 通用 Spark 支撑一系列函数库和软件栈，包括 SQL、DataFrames、机器学习 MLlib、GraphX 和 Spark Streaming，可以将这些库无缝地集成到同一个应用，将 SQL、流以及复杂分析结合在一起。
 - 运行在任何地方 Spark 可以运行在 Hadoop、Mesos、standalone 模式或云。它可以访问多种多样的数据，包括 HDFS、Cassandra、HBase 和 S3。可以以不同模式运行 Spark，包括本地模式、Standalone 模式、Mesoes 模式和 yarn 模式。
IBM Spectrum Conductor with Spark 能够简化开源大数据分析平台 Apache Spark 的部署，将其分析速度提升近 60%。作为一种开源大数据分析框架，Apache Spark 提供令人信服的性能优势。 实施 Spark 极具挑战性，包括投资新的专业能力、工具和工作流等。设置临时 Spark 集群可能导致无法高效利用资源，并带来管理和安全挑战。IBM Spectrum Conductor with Spark 可帮助解决这些问题。它将 Spark 发行与资源、基础架构和数据生命周期管理集成，以精简的方式创建企业级多租户 Spark 环境。为了帮助管理快速变迁的 Spark 生命周期，IBM Spectrum Conductor with Spark 支持同 时运行 Spark 的多种实例和版本。
本文档所做的测试，是基于 IBM Conductor with Spark 架构，它包含 3 台 Firestone 服务器，1 台 Driver 节点，2 台 Worker 节点，即 1+2 结构。如图 6 所示：
 图 6、Conductor with spark 集群架构
运行环境是 Conductor with spark, Spark 版本是 1.6.1, 采用 Spark 缺省的 DAGScheduler 调度软件，运行 gatk-launch 的选项是 --sparkRunner SPARK --sparkMaster spark://c712f6n10:7077。
4. 利用 Spark 技术进行基因序列分析处理 以 ReadsPipelineSpark 工作流为例，它是 GATK4 预定义的一个工作流，用 BAM 文件为输入，运行 MarkDuplicate 和 BQSR ，其输出文件将用于下一阶段分析。
重复是指一组基因片段有相同的、未修饰的起始和结束，MarkDuplicate 就是要挑选出"最佳的"复制，从而减缓错误效应。
 - BQSR 对一个已经排序过的 BAM 文件的合成测序数据基础质量数值进行重新调整，重新调整后，在 BAM 输出中每个片段在 QUAL 域中更精确，其报告的质量数值更接近于参考基因的实际可能性。
 运行命令：
清单 1.1
2
3
4
5
6
7
8
9
10
11
12
13
| ./gatk/gatk-launch \
ReadsPipelineSpark \ # Pipeline name
-I $bam \ # Input file
-R $ref \ # Reference file
-O $bamout \ # Output file
–bamPartitionSize 134217728 \ # maximum number of bytes to read from a file into each partition of reads.
–knownSites $dbsnp \ # knownSites (see notes)
–shardedOutput true \ # Write output to multiple pieces
–duplicates_scoring_strategy # MarkDuplicatesScoringStrategy
SUM_OF_BASE_QUALITIES \
–sparkRunner SPARK \ # Run mode
–sparkMaster spark://c712f6n10:7077 # Spark cluster
--conf spark.driver.memory=5g --conf spark.executor.memory=16g
|
输入文件：
1
2
3
| -I CEUTrio.HiSeq.WEx.b37.NA12892.bam
-R human_g1k_v37.2bit
-knownSites dbsnp_138.b37.excluding_sites_after_129.vcf
|
利用不同的资源管理器，还可以定义–num-executors, --executor-mem, --executor-cores，从而根据计算资源的大小合理分配和调度资源。
5. 性能和加速比分析 本文档以 CountReadsSpark 为例，对比分析在不同模式下的运行结果。
- 单机、非 Spark 模式下运行 CountReads:
清单 2.1
2
3
4
5
6
7
8
| #./gatk-launch CountReads -I /home/dlspark/SRR034975.Sort_all.bam
Running:
/home/dlspark/gatk/build/install/gatk/bin/gatk CountReads -I /home/dlspark/SRR034975.Sort_all.bam
[May 31, 2016 9:52:01 PM EDT] org.broadinstitute.hellbender.tools.CountReads --input /home/dlspark/SRR034975.Sort_all.bam --disable_all_read_filters false --interval_set_rule UNION --interval_padding 0 --readValidationStringency SILENT --secondsBetweenProgressUpdates 10.0 --disableSequenceDictionaryValidation false --createOutputBamIndex true --createOutputBamMD5 false --addOutputSAMProgramRecord true --help false --version false --verbosity INFO --QUIET false
Output: 34929382
Elapsed time: 12.14 minutes
|
- 在 Spark Standalone 模式下运行 CountReadsSpark
清单 3.1
2
3
4
5
6
7
| # ./gatk-launch CountReadsSpark -I /home/dlspark/SRR034975.Sort_all.bam
Running:
/home/dlspark/gatk/build/install/gatk/bin/gatk CountReadsSpark -I /home/dlspark/SRR034975.Sort_all.bam
[June 1, 2016 1:11:34 AM EDT] org.broadinstitute.hellbender.tools.spark.pipelines.CountReadsSpark --input /home/dlspark/SRR034975.Sort_all.bam --readValidationStringency SILENT --interval_set_rule UNION --interval_padding 0 --bamPartitionSize 0 --disableSequenceDictionaryValidation false --shardedOutput false --numReducers 0 --sparkMaster local --help false --version false --verbosity INFO --QUIET false
Output: 34929382
Elapsed time: 9.50 minutes
|
- 在 Spark Cluster 模式下运行 CountReadsSpark
清单 4.1
2
3
4
5
6
| # ./gatk-launch CountReadsSpark -I /gpfs1/yrx/SRR034975.Sort_all.bam -O /gpfs1/yrx/gatk4-test.output --sparkRunner SPARK --sparkMaster spark://c712f6n10:7077
Running:
spark-submit --master spark://c712f6n10:7077 --conf spark.kryoserializer.buffer.max=512m --conf spark.driver.maxResultSize=0 --conf spark.driver.userClassPathFirst=true --conf spark.io.compression.codec=lzf --conf spark.yarn.executor.memoryOverhead=600 --conf spark.yarn.dist.files=/home/dlspark/gatk/build/libIntelDeflater.so --conf spark.driver.extraJavaOptions=-Dsamjdk.intel_deflater_so_path=libIntelDeflater.so -Dsamjdk.compression_level=1 -DGATK_STACKTRACE_ON_USER_EXCEPTION=true --conf spark.executor.extraJavaOptions=-Dsamjdk.intel_deflater_so_path=libIntelDeflater.so -Dsamjdk.compression_level=1 -DGATK_STACKTRACE_ON_USER_EXCEPTION=true
Output: 34929382
Elapsed time: 0.60 minutes
|
图 7 是三种不同运算模式结果的比较，从运行时间看，在 Spark Cluster 模式下，CountReadsSpark 的运行效率比 Spark Standalone 模式提高了 15 倍。
 图 7、3 种运算模式结果比较
总结 通过以上的分析，基于 Spark 技术的生命科学解决方案，能够使原来串行的应用在不修改代码或修改少量代码的情况下就可以实现并行化、内存内计算，或使用几条 Java 代码即可在集群上处理大量的数据，不需要复杂的 MPI 或 OpenMP 编程，使科学家将精力更多集中于新方法的研究和新的发现。随着基于 Spark 的生命科学解决方案的不断成熟和完善，越来越多的工具和工作流可以运行在 Spark 计算平台，它具有容错功能，扩展性也不断提高，并行化的生命科学应用将会使序列分析时间从目前的十几个小时缩短到 1 个小时之内，同时利用分布式作业调度和低成本集群，也会极大地降低分析成本。 |
