【DataMagic】如何在万亿级别规模的数据量上使用Spark

欢迎你们前往腾讯云+社区，获取更多腾讯海量技术实践干货哦~

本文由鹅厂新鲜事儿发表于云+社区专栏

做者：张国鹏 | 腾讯 运营开发工程师

1、前言

Spark做为大数据计算引擎，凭借其快速、稳定、简易等特色，快速的占领了大数据计算的领域。本文主要为做者在搭建使用计算平台的过程当中，对于Spark的理解，但愿能给读者一些学习的思路。文章内容为介绍Spark在DataMagic平台扮演的角色、如何快速掌握Spark以及DataMagic平台是如何使用好Spark的。

2、Spark在DataMagic平台中的角色

图 2-1

整套架构的主要功能为日志接入、查询(实时和离线)、计算。离线计算平台主要负责计算这一部分，系统的存储用的是COS(公司内部存储)，而非HDFS。

下面将主要介绍Spark on Yarn这一架构，抽取出来即图2-2所示，能够看到Spark on yarn的运行流程。

图2-2

3、如何快速掌握Spark

对于理解Spark，我以为掌握下面4个步骤就能够了。

1．理解Spark术语

对于入门，学习Spark能够经过其架构图，快速了解其关键术语，掌握了关键术语，对Spark基本上就有认识了，分别是结构术语Shuffle、Patitions、MapReduce、Driver、Application Master、Container、Resource Manager、Node Manager等。API编程术语关键RDD、DataFrame，结构术语用于了解其运行原理，API术语用于使用过程当中编写代码，掌握了这些术语以及背后的知识，你就也知道Spark的运行原理和如何编程了。

2．掌握关键配置

Spark在运行的时候，不少运行信息是经过配置文件读取的，通常在spark-defaults.conf，要把Spark使用好，须要掌握一些关键配置，例如跟运行内存相关的，spark.yarn.executor.memoryOverhead、spark.executor.memory，跟超时相关的spark.network.timeout等等，Spark不少信息均可以经过配置进行更改，所以对于配置须要有必定的掌握。可是使用配置时，也要根据不一样的场景，这个举个例子，例如spark.speculation配置，这个配置主要目的是推测执行，当worker1执行慢的状况下，Spark会启动一个worker2，跟worker1执行相同的任务，谁先执行完就用谁的结果，从而加快计算速度，这个特性在通常计算任务来讲是很是好的，可是若是是执行一个出库到Mysql的任务时，同时有两个同样的worker，则会致使Mysql的数据重复。所以咱们在使用配置时，必定要理解清楚，直接google spark conf就会列出不少配置了。

3．使用好Spark的并行

咱们之因此使用Spark进行计算，缘由就是由于它计算快，可是它快的缘由很大在于它的并行度，掌握Spark是如何提供并行服务的，从而是咱们更好的提升并行度。

对于提升并行度，对于RDD，须要从几个方面入手，一、配置num-executor。二、配置executor-cores。三、配置spark.default.parallelism。三者之间的关系通常为spark.default.parallelism=num-executors*executor-cores的2~3倍较为合适。对于Spark-sql，则设置spark.sql.shuffle.partitions、num-executor和executor-cores。

4．学会如何修改Spark代码

新手而言，特别是须要对Spark进行优化或者修改时，感到很迷茫，其实咱们能够首先聚焦于局部，而Spark确实也是模块化的，不须要以为Spark复杂而且难以理解，我将从修改Spark代码的某一角度来进行分析。

首先，Spark的目录结构如图3-1所示，能够经过文件夹，快速知道sql、graphx等代码所在位置，而Spark的运行环境主要由jar包支撑，如图3-2所示，这里截取部分jar包，实际上远比这多，全部的jar包均可以经过Spark的源代码进行编译，当须要修改某个功能时，仅须要找到相应jar包的代码，修改以后，编译该jar包，而后进行替换就好了。

图3-1

图3-2

而对于编译源代码这块，其实也很是简单，安装好maven、scala等相关依赖，下载源代码进行编译便可，掌握修改源码技巧对于使用好开源项目十分重要。

4、DataMagic平台中的Spark

Spark在DataMagic中使用，也是在边使用边探索的过程，在这过程当中，列举了其比较重要的特色。

1．快速部署

在计算中，计算任务的数量以及数据的量级天天都会发生变化，所以对于Spark平台，须要有快速部署的特性，在实体机上，有一键部署脚本，只要运行一个脚本，则能够立刻上线一个拥有128G内存、48cores的实体机，可是实体机一般须要申请报备才能得到，所以还会有docker来支持计算资源。

2．巧用配置优化计算

Spark大多数属性都是经过配置来实现的，所以能够经过配置动态修改Spark的运行行为，这里举个例子，例如经过配置自动调整exector的数量。

2.1 在nodeManager的yarn-site.xml添加配置

<property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce_shuffle,spark_shuffle</value>
   </property>
   <property>
      <name>yarn.nodemanager.aux-services.spark_shuffle.class</name>
      <value>org.apache.spark.network.yarn.YarnShuffleService</value>
   </property>
复制代码

2.2 将spark-2.2.0-yarn-shuffle.jar文件拷贝到hadoop-yarn/lib目录下（即yarn的库目录）

2.3 在Spark的spark-default.xml添加配置

spark.dynamicAllocation.minExecutors 1 #最小Executor数
spark.dynamicAllocation.maxExecutors 100 #最大Executor数
复制代码

经过这种配置，能够达到自动调整exector的目的。

3．合理分配资源

做为一个平台，其计算任务确定不是固定的，有的数据量多，有的数据量少，所以须要合理分配资源，例若有些千万、亿级别的数据，分配20核计算资源就足够了。可是有些数据量级达到百亿的，就须要分配更多的计算资源了。参考第三章节的第3点。

4．贴合业务需求

计算的目的其实就是为了服务业务，业务的需求也理应是平台的追求，当业务产生合理需求时，平台方也应该尽可能去知足。如为了支持业务高并发、高实时性查询的需求下，Spark在数据出库方式上，支持了Cmongo的出库方式。

sc = SparkContext(conf=conf) sqlContext = SQLContext(sc) database = d = dict((l.split('=') for l in dbparameter.split())) parquetFile = sqlContext.read.parquet(file_name) parquetFile.registerTempTable(tempTable) result = sqlContext.sql(sparksql) url = "mongodb://"+database['user']+":"+database['password']+"@"+database['host']+":"+database['port'] result.write.format("com.mongodb.spark.sql").mode('overwrite').options(uri=url,database=database['dbname'],collection=pg_table_name).save()

5．适用场景

Spark做为通用的计算平台，在普通的应用的场景下，通常而言是不须要额外修改的，可是DataMagic平台上，咱们须要“在前行中改变”。这里举个简单的场景，在日志分析中，日志的量级达到千亿/日的级别，当底层日志的某些字段出现utf-8编码都解析不了的时候，在Spark任务中进行计算会发生异常，而后失败，然而若是在数据落地以前对乱码数据进行过滤，则有可能会影响数据采集的效率，所以最终决定在Spark计算过程当中解决中这个问题，所以在Spark计算时，对数据进行转换的代码处加上异常判断来解决该问题。

6．Job问题定位

Spark在计算任务失败时候，须要去定位失败缘由，当Job失败是，能够经过yarn logs -applicationId application 来合并任务log，打开log，定位到Traceback，通常能够找到失败缘由。通常而言，失败能够分红几类。

a. 代码问题，写的Sql有语法问题，或者Spark代码有问题。

b. Spark问题，旧Spark版本处理NULL值等。

c. 任务长时间Running状态，则多是数据倾斜问题。

d. 任务内存越界问题。

7．集群管理

Spark集群在平常使用中，也是须要运营维护的，从而运营维护，发现其存在的问题，不断的对集群进行优化，这里从如下几个方面进行介绍，经过运营手段来保障集群的健壮性和稳定性，保证任务顺利执行。

a. 定时查看是否有lost node和unhealthy node，能够经过脚原本定时设置告警，若存在，则须要进行定位处理。

b. 定时扫描hdfs的运行log是否满了，须要定时删除过时log。

c. 定时扫描集群资源是否知足计算任务使用，可以提早部署资源。

5、总结

本文主要是经过做者在搭建使用计算平台的过程当中，写出对于Spark的理解，而且介绍了Spark在当前的DataMagic是如何使用的，当前平台已经用于架平离线分析，天天计算分析的数据量已经达到千亿~万亿级别。

问答

Spark 优化怎么作？

相关阅读

腾讯云游戏多媒体解决方案

害怕羊毛党？腾讯云为你保驾护航

AI从入门到放弃：CNN的导火索，用MLP作图像分类识别？

【每日课程推荐】机器学习实战！快速入门在线广告业务及CTR相应知识

此文已由做者受权腾讯云+社区发布，更多原文请点击

搜索关注公众号「云加社区」，第一时间获取技术干货，关注后回复1024 送你一份技术课程大礼包！

海量技术实践经验，尽在云加社区！