一. MaxCompute Spark 介绍
MaxCompute Spark是MaxCompute提供的兼容开源的Spark计算服务。它在统一的计算资源和数据集权限体系之上,提供Spark计算框架,支持用户以熟悉的开发使用方式提交运行Spark作业,以满足更丰富的数据处理分析场景。
1.1 关键特性
- 支持原生多版本Spark作业
- 社区原生Spark运行在MaxCompute里,完全兼容Spark的API,支持多个Spark版本同时运行
- 统一的计算资源
- 像MaxCompute SQL/MR等任务类型一样,运行在MaxCompute项目开通的统一计算资源中
- 统一的数据和权限管理
- 遵循MaxCompute项目的权限体系,在访问用户权限范围内安全地查询数据
- 与开源系统相同的使用体验
- 提供原生的开源实时Spark UI和查询历史日志的功能
1.2 系统结构
- 原生Spark通过MaxCompute Cupid平台能够在MaxCompute中运行
Spark
Spark
Spark
Spark
Spark
Spark
Spark
Spark
M1ib
Streaming
SQL
Graphx
Mlib
Streaming
Graphx
SQL
SparkCore
SparkCore
Cupid
Yarn/Mesos
HDFS
HDFSInterface
YarnInterface
Fuxi
Pangu
1.3 约束与限制
- 目前MaxCompute Spark支持以下适用场景:
- 离线计算场景:GraphX、Mllib、RDD、Spark-SQL、PySpark等
- Streaming场景
- 读写MaxCompute Table
- 引用MaxCompute中的文件资源
- 读写VPC环境下的服务,如RDS、Redis、HBase、ECS上部署的服务等
- 读写OSS非结构化存储
- 使用限制
- 不支持交互式类需求Spark-Shell、Spark-SQL-Shell、PySpark-Shell等
- 不支持访问MaxCompute外部表,函数和UDF
- 只支持Local模式和Yarn-cluster模式运行
二. 开发环境搭建
2.1 运行模式
- 通过Spark客户端提交
- Yarn-Cluster模式,提交任务到MaxCompute集群中
- Local模式
- 通过Dataworks提交
- 本质上也是Yarn-Cluster模式,提交任务到MaxCompute集群中
2.2 通过客户端提交
2.2.1 Yarn-Cluster模式
- 下载MC Spark客户端
- Spark 1.6.3
- Spark 2.3.0
- 环境变量配置
## JAVA_HOME配置 # 推荐使用JDK 1.8 export JAVA_HOME=/path/to/jdk export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar export PATH=$JAVA_HOME/bin:$PATH ## SPARK_HOME设置 # 下载上文提到的MaxCompute Spark客户端并解压到本地任意路径 # 请不要直接设置SPARK_HOME等于以下路径下述路径仅做展示用途 # 请指向正确的路径 export SPARK_HOME=/path/to/spark_extracted_package export PATH=$SPARK_HOME/bin:$PATH ## PySpark配置Python版本 export PATH=/path/to/python/bin/:$PATH
- 参数配置
- 将$SPARK_HOME/conf/spark-defaults.conf.template 重命名为 spark-defaults.conf
- 参数配置参考下文
- 准备项目工程
git clone https://github.com/aliyun/MaxCompute-Spark.git cd spark-2.x mvn clean package
- 任务提交
// bash环境 cd $SPARK_HOME bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.SparkPi \ /path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar // 在windows环境提交的命令 cd $SPARK_HOME/bin spark-submit.cmd --master yarn-cluster --class com.aliyun.odps.spark.examples.SparkPi \path\to\MaxCompute-Spark\spark-2.x\target\spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar
2.2.2 Local模式
- 与Yarn Cluster模式类似,用户首先需要做以上准备工作
- 任务提交
## Java/Scala cd $SPARK_HOME ./bin/spark-submit --master local[4] --class com.aliyun.odps.spark.examples.SparkPi \ /path/to/odps-spark-examples/spark-examples/target/spark-examples-2.0.0-SNAPSHOT-shaded.jar ## PySpark cd $SPARK_HOME ./bin/spark-submit --master local[4] \ /path/to/odps-spark-examples/spark-examples/src/main/python/odps_table_rw.py
- IDEA调试注意
- IDEA运行Local模式是不能直接引用spark-defaults.conf里的配置,需要手动在代码里指定相关配置
- 一定要注意需要在IDEA里手动添加MaxCompute Spark客户端的相关依赖(jars目录),否则会出现以下报错:
the value of spark.sql.catalogimplementation should be one of hive in-memory but was odps
- 参考文档
2.3 通过DataWorks提交
2.3.1 资源上传
- 本质上MC Spark节点的配置对应于spark-submit命令的参数和选项
ODPS SPARK节点 | spark-submit |
主Java、Python资源 | app jar or python file |
配置项 | --conf PROP=VALUE |
Main Class | --class CLASS_NAME |
参数 | [app arguments] |
选择JAR资源 | --jars JARS |
选择Python资源 | --py-files PY_FILES |
选择File资源 | --files FILES |
选择Archives资源 | --archives |
zky_shell
Test
Leanken
spark-examples_211-1.0.
spark-examples2.11-3.3..-
dytest
ODPSSpark
Spark1.x
Spark2.x
spark藏本:
Pyithon
JavavScala
选择主jor资源:
spattexamples211-1.0.0SMAPSHOTHadeda
配置项:
Trkexecutor.instance8
利
利除
利脸
Sparkdrwver.cores
丑(脸
sparkdriver.memory
漂脑
http://serwicecnmaxcompurtealyun.com/api
sparkhadopp.odps.endpolnt
丑族
sparkhadoop.odpsnntimeendpoint
httP//Sericecnmaxcomputealiyuninc.com/api
利除
cupid_v2
sparkhadoop.odps.task.majorversion
odps
spark.sql.cataloglmplementation
清构
sparkzhadoop.odps.cupid.container.imageenable
Sparthadoop.odps.cupidcontainerengie.tpe
洲除
hyper
语加一条
MainClass
com.atyunodpssparkexamples.Spar
梦热:
多个参鼓之间用空格分限
逸择ir涛:
请选择
选择价le梵源:
请选择
选择archwes资源:
请远择
- 上传资源:
- 0~50MB:可以直接在DataWorks界面创建资源并上传
- 50MB~500MB:可以先利用MaxCompute客户端(CMD)上传,然后在DataWorks界面添加到数据开发,参考文档
- 资源引用:
- 资源提交后,可以在DataWorks Spark节点界面选择需要的资源(jar/python/file/archive)
- 任务运行时:资源文件默认会上传到Driver和Executor的当前工作目录
2.3.2 参数和配置
- Spark 配置项:对应于spark-submit命令的--conf选项
- accessid,accesskey,projectname,endpoint,runtime.end.point,task.major.version无需配置
- 除此之外,需要将spark-default.conf中的配置逐条加到dataworks的配置项中
- 给主类传参数(如bizdate)
- 首先在调度->参数中添加参数,然后在Spark节点“参数”栏引用该参数。多个参数用空格分隔
- 该参数会传给用户主类,用户在代码中解析该参数即可
- 参考文档
三. 配置介绍
3.1 配置的位置
3.1.1 Spark配置的位置
- 用户使用Maxcompute Spark通常会有几个位置可以添加Spark配置,主要包括:
- 位置1:spark-defaults.conf,用户通过客户端提交时在spark-defaults.conf文件中添加的Spark配置
- 位置2:dataworks的配置项,用户通过dataworks提交时在配置项添加的Spark配置,这部分配置最终会在位置3中被添加
- 位置3:配置在启动脚本spark-submit --conf选项中
- 位置4:配置在用户代码中,用户在初始化SparkContext时设置的Spark配置
- Spark配置的优先级
- 用户代码 > spark-submit --选项 > spark-defaults.conf配置 > spark-env.sh配置 > 默认值
3.1.2 需要区分的两种配置
- 一种是必须要配置在spark-defaults.conf或者dataworks的配置项中才能生效(在任务提交之前需要),而不能配置在用户代码中,这类配置主要的特征是:
- 与Maxcompute/Cupid平台相关:一般参数名中都会带odps或者cupid,通常这些参数与任务提交/资源申请都关系:
- 显而易见,一些资源获取(如driver的内存,core,diskdriver,maxcompute资源),在任务执行之前就会用到,如果这些参数设置在代码中,很明显平台没有办法读到,所以这些参数一定不要配置在代码中
- 其中一些参数即使配置在代码中,也不会造成任务失败,但是不会生效
- 其中一些参数配置在代码中,可能会造成副作用:如在yarn-cluster模式下设置spark.master为local
- 访问VPC的参数:
- 这类参数也与平台相关,打通网络是在提交任务时就进行的
- 一种是在以上三种位置配置都可以生效,但是在代码配置的优先级最高
- 推荐把任务运行与优化相关的参数配置在代码中,而与资源平台相关的配置都配置在spark-defaults.conf或者dataworks的配置项中。
3.2 资源相关的参数
spark.executor.instances |
|
spark.executor.cores |
|
spark.executor.memory |
|
spark.yarn.executor.memoryOverhead |
|
spark.driver.cores |
|
spark.driver.memory |
|
spark.yarn.driver.memoryOverhead |
|
spark.driver.maxResultSize |
|
spark.hadoop.odps.cupid.disk.driver.device_size |
|
3.3 平台相关的参数
spark.hadoop.odps.project.name |
|
spark.hadoop.odps.access.id |
|
spark.hadoop.odps.access.key |
|
spark.hadoop.odps.end.point |
|
spark.hadoop.odps.runtime.end.point |
|
spark.hadoop.odps.task.major.version |
|
spark.sql.catalogImplementation |
|
spark.hadoop.odps.cupid.resources |
|
spark.hadoop.odps.cupid.vectorization.enable |
|
spark.hadoop.odps.input.split.size |
|
spark.hadoop.odps.cupid.vpc.domain.list |
|
spark.hadoop.odps.cupid.smartnat.enable |
|
spark.hadoop.odps.cupid.eni.enable |
|
spark.hadoop.odps.cupid.eni.info |
|
spark.hadoop.odps.cupid.engine.running.type |
|
spark.hadoop.odps.cupid.job.capability.duration.hours |
|
spark.hadoop.odps.moye.trackurl.dutation |
|
四. 作业诊断
4.1 Logview
4.1.1 Logview 介绍
- 在任务提交时会打印日志: 日志中含有logview链接 (关键字 logview url)
- Master以及Worker的StdErr打印的是spark引擎输出的日志,StdOut中打印用户作业输出到控制台的内容
4.1.2 利用Logview 排查问题
- 拿到Logview,一般首先看Driver的报错,Driver会包含一些关键性的错误
- 如果Driver中出现类或者方法找不到的问题,一般是jar包打包的问题
- 如果Driver中出现连接外部VPC或者OSS出现Time out,这种情况一般要去排查一下参数配置
- 如果Driver中出现连接不到Executor,或者找不到Chunk等错误,通常是Executor已经提前退出,需要进一步查看Executor的报错,可能存在OOM
- 根据End Time做排序,结束时间越早,越容易是发生问题的Executor节点
- 根据Latency做排序,Latency代表了Executor的存活的时间,存活时间越短的,越有可能是根因所在
atency:min:00:001
ILong-Tails(1)
SmartFilter
Terminated
Data-Skews
FAIled
EndTIme
V/OBytes
V/oRecord
StartTIme
Status
StdEr
Stdout
Debug
Progress
Latency
良
2020/11/0514:21:54
2020/11/0514:21:43
0B/oB
00:00:11
0/o
Terminated
00:00:20
0B/oB
2020/11/0514:12:16
2020/11/0514:11:56
Terminated
0/o
00:00:23
0B/oB
2020/11/0514:12:51
2020/11/0514:12:28
0/o
Terminated
100%
2020/11/0514:21:33
2020/11/0514:15:35
0/0
00:05:58
0B/OB
Torminated
100%
2020/11/0514:21:58
2020/11/0514:11:58
0/0
00:10:00
0B/OB
Terminated
100%
4.2 Spark UI和HistoryServer
- Spark UI与社区版一致,在logivew的summary模块下找到Spark UI链接:
SourceXML
SubS
JobDetails
History
Result
Summary
JsonSummary
SparkuI(JobstiltRunning)
http://20201119053910902g01imui.nmxcomuecm
dapplication6
+OKENVWSYZZ+RMWZRSTALAYCZLPLUPPSXPRFBTXO9CTZPWNF8YMDGSMDg5OC
kCHM6kjoqfiwgkZ
5
Cmd+clicktofollowlink
nkished):
http://spark-history-cn.oaxcomuteucomctn
i20201119053910902goontkidacon
16057643195071194698
TOKenWWYZ+ZZRMWZRSIYCLPUPPSXPRFBX9CTZPMNF8YMDGSMDG5OL
SKCHHGKQIWKVZ
- Spark UI的使用与社区原生版是一致的,可以参考文档
- 注意
- Spark UI需要鉴权,只有提交任务的Owner才能打开
- Spark UI仅在作业运行时才能打开,如果任务已经结束,那么Spark UI是无法打开的,这时候需要查看Spark History Server UI
五. 常见问题
1. local模式运行的问题
- 问题一:the value of spark.sql.catalogimplementation should be one of hive in-memory but was odps
- 原因在于用户没有正确地按照文档将Maxcompute Spark的jars目录添加到类路径,导致加载了社区版的spark包,需要按照文档将jars目录添加到类路径
- 问题二:IDEA Local模式是不能直接引用spark-defaults.conf里的配置,必须要把Spark配置项写在代码中
- 问题三:访问OSS和VPC:
- Local模式是处于用户本机环境,网络没有隔离。而Yarn-Cluster模式是处于Maxcompute的网络隔离环境中,必须要要配置vpc访问的相关参数
- Local模式下访问oss的endpoint通常是外网endpoint,而Yarn-cluster模式下访问vpc的endpoint是经典网络endpoint
2. jar包打包的问题
- java/scala程序经常会遇到Java类找不到/类冲突问题:
- 类冲突:用户Jar包与Spark或平台依赖的Jar包冲突
- 类没有找到:用户Jar包没有打成Fat Jar或者由于类冲突引起
- 打包需要注意:
- 依赖为provided和compile的区别:
- provided:代码依赖该jar包,但是只在编译的时候需要用,而运行时不需要,运行时会去集群中去寻找的相应的jar包
- compile:代码依赖该jar包,在编译、运行时候都需要,在集群中不存在这些jar包,需要用户打到自己的jar包中。这种类型的jar包一般是一些三方库,且与spark运行无关,与用户代码逻辑有关
- 用户提交的jar包必须是Fat jar:
- 必须要把compile类型的依赖都打到用户jar包中,保证代码运行时能加载到这些依赖的类
- 需要设置为provided的jar包
- groupId为org.apache.spark的Jar包
- 平台相关的Jar包
- cupid-sdk
- hadoop-yarn-client
- odps-sdk
- 需要设置为compile的jar包
- oss相关的jar包
- hadoop-fs-oss
- 用户访问其他服务用到的jar包:
- 如mysql,hbase
- 用户代码需要引用的第三方库
3. 需要引入Python包
- 很多时候用户需要用到外部Python依赖
- 首先推荐用户使用我们打包的公共资源,包含了常用的一些数据处理,计算,以及连接外部服务(mysql,redis,hbase)的三方库
## 公共资源python2.7.13 spark.hadoop.odps.cupid.resources = public.python-2.7.13-ucs4.tar.gz spark.pyspark.python = ./public.python-2.7.13-ucs4.tar.gz/python-2.7.13-ucs4/bin/python ## 公共资源python3.7.9 spark.hadoop.odps.cupid.resources = public.python-3.7.9-ucs4.tar.gz spark.pyspark.python = ./public.python-3.7.9-ucs4.tar.gz/python-3.7.9-ucs4/bin/python3
- 如果不能满足用户需要,用户可以在该公共资源的基础上上传wheel包
- 如果wheel包依赖链较为复杂,可以通过Docker容器进行打包
- 使用Docker容器打包:
- 为了保证与线上环境一致,避免运行时so包找不到的问题,需要使用Docker容器进行打包
- Docker容器本质只是提供了兼容性较好的os环境,用户需要在容器中进行打包,并将整个Python目录压缩后上传到MaxCompute Resource中,最后在Spark任务中直接引用即可
- 参见文档
4. 需要引入外部文件
- 需要引用到外部文件的场景
- 用户作业需要读取一些配置文件
- 用户作业需要额外的jar包/Python库
- 可以通过两种方式上传资源:
- 通过Spark参数上传文件
- 通过MaxCompute Resource上传文件
- 通过Spark参数上传文件
- MaxCompute Spark支持Spark社区版原生的--jars,--py-files等参数,可以在作业提交时通过这些参数将文件上传,这些文件在任务运行时会被上传到用户的工作目录下
- 通过DataWorks添加任务需要的资源,参见上文
- MaxCompute Resource
- spark.hadoop.odps.cupid.resources参数,可以直接引用MaxCompute中的资源,这些资源在任务运行时也会被上传到用户的工作目录下
- 使用方式
(1)通过MaxCompute客户端将文件上传(单个文件最大支持500MB)
(2)在Spark作业配置中添加spark.hadoop.odps.cupid.resources参数:格式为<projectname>.<resourcename>,如果需要引用多个文件,需要用逗号隔开
(3)如果需要重命名,格式为<projectname>.<resourcename>:<new resource name>
- 如何读取上传的文件:
- 如果需要读取上传的文件资源,文件路径如下:
val dir = new File(".") val targetFile = "file://" + dir.getCanonicalPath + "/" +文件名
- 或者直接通过类加载器获取文件路径,然后再读取
- 参考文档
5. VPC访问的问题
- Maxcompute Spark是独立运行在Maxcompute集群的,网络与外界隔离,因此无法直接访问vpc和公网,需要添加以下配置。
- 北京和上海Region使用smartnat
- 需要配置
- spark.hadoop.odps.cupid.vpc.domain.list
- spark.hadoop.odps.cupid.smartnat.enable=true
- 访问公网:假如要访问google.com:443,需要做以下两步:
- 提工单设置 project 级别白名单,把 google.com:443 加到odps.security.outbound.internetlist
- 配置作业级别的公网访问白名单:spark.hadoop.odps.cupid.internet.access.list=google.com:443
- 其他Region:
- 只需要配置spark.hadoop.odps.cupid.vpc.domain.list
- 无法访问公网
- 注意事项:
- vpc.domain.list 需要压缩成一行,不能包含空格
- 支持同时访问同一个Region下的多个VPC,需要配置所有要访问的ip:port的白名单
- 需要在要访问的服务中添加ip白名单,允许100.104.0.0/16网段的访问
- 用户要保证所有可能访问到的IP都已经加到vpc.domain.list,例如如果用户要访问位于hdfs,hbase这种多个节点的服务,一定要把所有的节点都添加进来,不然可能会遇到Time out
6. OOM的问题
- 可能出现OOM的情况:
- 错误1: 在某些Executor中出现Cannot allocate memory,一般是系统内存不足,此时可以调整spark.yarn.executor.memoryOverhead参数,注意该参数是会计算到总内存数的,也不需要一次性增加太多,小心调整即可
- 错误2:Executor抛出java.lang.OutOfMemoryError: Java heap space
- 错误3:GC overhead limit exceeded
- 错误4:No route to host: workerd*********/Could not find CoarseGrainedScheduler,这类错误一般是一些Executor提前退出。如果一个task处理的数据非常大,容易发生OOM
- Driver OOM:Driver OOM的可能性比较小,但是也是有可能出现的
- 如果需要使用collect算子将RDD的数据全部拉取到Driver上进行处理,那么必须确保Driver的内存足够大,否则会出现OOM内存溢出的问题。
- SparkContext,DAGScheduler都是运行在Driver端的。Stage切分也是在Driver端运行,如果用户程序有过多的步骤,切分出过多的Stage,这部分信息消耗的是Driver的内存,这个时候就需要调大Driver的内存。有时候如果stage过多,Driver端可能会有栈溢出的问题
- 一些解决方法:
- 限制executor 并行度,将cores 调小:多个同时运行的 Task 会共享一个Executor 的内存,使得单个 Task 可使用的内存减少,调小并行度能缓解内存压力
- 增加单个Executor内存
- 增加分区数量,减少每个executor负载
- 考虑数据倾斜问题,因为数据倾斜导致某个 task 内存不足,其它 task 内存足够
7. No space left on device
- 这个错误意味这本地磁盘不足,通常这个报错会在executor上出现,并导致executor挂掉
- 解决方案
- 直接增加更多的磁盘空间:默认driver和executor都各提供20g的本地磁盘,当磁盘空间不足时可以调整spark.hadoop.odps.cupid.disk.driver.device_size
- 如果调整本地磁盘大小到100g后,仍然报该错误,说明单个executor写的shuffle数据已经超过上限,可能是遇到了数据倾斜,这种情况下可以对数据重分区。或者增加executor的数量
8. 申请资源的问题
- 申请不到资源的几种现象:
(1)在driver端一般会打以下日志
- WARN YarnClusterScheduler: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources
(2)在logview中只能看到driver,而worker数量为0
(3)在spark ui中只能看到driver,而worker数量为0
- 解决方案:
- 调整任务资源:调整用户申请的executor总数或者单个executor的资源数量(一般是内存),如果单个executor请求的内存过多可能不太容易申请到
- 合理安排任务执行时间
- 其他注意事项:
- 必须配置spark.master=yarn-cluster才会正确的申请资源
9. 其他问题
- 如何切换Spark版本
- 版本号规则介绍:示例spark-2.3.0-odps0.32.5
- spark-2.3.0是社区版本的spark版本号,Maxcompute Spark基于该社区版本进行适配
- odps0.32.5是Maxcompute Spark的小版本号,随着小版本号的升级,可能进行一些bug修复和sdk的升级
- 用户提交作业的的Spark版本可能有以下几种情况:
- 情况1:直接通过本地客户端提交任务,spark版本就是用户本地客户端的版本
- 情况2:用户通过dataworks提交任务,取决于dataworks gateway的默认spark版本,当前公共云dataworks 公共资源组gateway的默认版本是spark-2.3.0-odps0.32.1
- 情况3:用户通过dataworks提交任务,配置参数spark.hadoop.odps.spark.version,则会按照配置的版本号来寻找对应的spark客户端,用户可以配置spark.hadoop.odps.spark.version=spark-2.3.0-odps0.32.5手动切换版本
- 情况4:该情况优先级最高,用户可以在本地客户端或者是dataworks提交任务时配置以下参数,则类加载的优先级最高,因此会在spark任务启动时优先使用该版本的spark
spark.hadoop.odps.cupid.resources = public.__spark_libs__2.3.0odps0.32.5.zip spark.driver.extraClassPath = ./public.__spark_libs__2.3.0odps0.32.5.zip/* spark.executor.extraClassPath = ./public.__spark_libs__2.3.0odps0.32.5.zip/*
- 需要在代码中访问配置项:
- spark开头的参数直接通过SparkConf类提供的接口直接读取即可
- Spark History Server渲染速度慢
- 可以添加压缩配置:spark.eventLog.compress=true
- 如何正确地Kill一个运行中的Spark任务
- 通常通过两种方式kill正在运行的Spark任务
(1)通过odps cmd 执行 kill + instanceId;
(2)通过dataworks界面执行stop
- 注意,直接在spark客户端或者dataworks的任务提交界面执行Ctrl + C是无法kill一个Spark任务的
- 日志中文乱码,添加以下配置
- spark.executor.extraJavaOptions=-Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8
- spark.driver.extraJavaOptions=-Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8
- 如果是pyspark作业需要设置下如下两个参数:
- spark.yarn.appMasterEnv.PYTHONIOENCODING=utf8
- spark.executorEnv.PYTHONIOENCODING=utf8
- 另外在python脚本的最前面加上如下的代码:
# -*- coding: utf-8 -*- import sys reload(sys) sys.setdefaultencoding('utf-8')
六. 相关文档
- MC Spark github wiki:https://github.com/aliyun/MaxCompute-Spark/wiki
- Spark UI:https://spark.apache.org/docs/latest/web-ui.html
- Spark 配置:https://spark.apache.org/docs/2.4.5/configuration.html