MaxCompute Spark 使用和常见问题

一. 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中运行

       MaxCompute Spark 使用和常见问题

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


MaxCompute Spark 使用和常见问题

  • 上传资源:
    • 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

  • 总共申请的executor数目
  • 普通任务十几个或者几十个足够了,若是处理大量数据时可以申请多一些,100—2000+
spark.executor.cores
  • 每个executor的核数
  • Job的最大并行度是executor数目*executor core数

spark.executor.memory

  • 代表申请executor的内存

spark.yarn.executor.memoryOverhead

  • 申请executor的堆外内存,默认单位是MB
  • 主要用于JVM自身,字符串, NIO Buffer等开销
  • 单个executor的总内存是:spark.executor.memory+spark.yarn.executor.memoryOverhead

spark.driver.cores

  • 类似executor

spark.driver.memory

  • 类似executor

spark.yarn.driver.memoryOverhead

  • 类似executor

spark.driver.maxResultSize

  • 默认1g,控制worker送回driver的数据大小,一旦超过该限制,driver会终止执行

spark.hadoop.odps.cupid.disk.driver.device_size


  • 代表本地网盘大小,默认值为20g
  • 当出现No space left on device时可适当调大该值,最大支持100g
  • 设置该参数需要包含单位‘g’


3.3 平台相关的参数

spark.hadoop.odps.project.name

  • Spark任务运行所在的project

spark.hadoop.odps.access.id

  • 提交spark任务的accessId

spark.hadoop.odps.access.key

  • 提交spark任务的accessKey

spark.hadoop.odps.end.point

  • 用于任务提交
  • 中国公共云一般设置为

http://service.cn.maxcompute.aliyun.com/api

spark.hadoop.odps.runtime.end.point

  • 用于任务运行
  • 中国公共云一般设置为

http://service.cn.maxcompute.aliyun-inc.com/api

spark.hadoop.odps.task.major.version

  • 代表当前使用的平台版本
  • 公共云设置为cupid_v2即可

spark.sql.catalogImplementation

  • Spark 2.3版本需要设置为odps
  • Spark 2.4以后版本会改为hive
  • 为了便于作业迁移,建议不要将该配置写在代码中

spark.hadoop.odps.cupid.resources

  • 该配置项指定了程序运行所需要的Maxcompute资源,格式为<projectname>.<resourcename>,可指定多个,用逗号分隔。
  • 指定的资源将被下载到driver和executor的工作目录,经常使用该参数来引用较大的文件。
  • 资源下载到目录后默认的名字是<projectname>.<resourcename>
  • 如果需要重新命名,需要在配置时通过<projectname>.<resourcename>:<new resource name>进行重命名

spark.hadoop.odps.cupid.vectorization.enable

  • 是否开启向量化读写,默认为true

spark.hadoop.odps.input.split.size

  • 用于调节读Maxcompute表的并发度
  • 默认每个分区为256MB,该参数单位为MB

spark.hadoop.odps.cupid.vpc.domain.list

  • vpc访问依赖的参数,传统的访问vpc的方式

spark.hadoop.odps.cupid.smartnat.enable

  • vpc访问依赖的参数
  • 如果region是北京或者上海,将该参数设置为true

spark.hadoop.odps.cupid.eni.enable

  • 如果用户已开通专线,则需要配置为true

spark.hadoop.odps.cupid.eni.info

  • 如果用户已开通专线,则需要设置该参数
  • 该参数代表用户打通的vpc

spark.hadoop.odps.cupid.engine.running.type

  • 普通作业3天没跑完就会被强制回收,流式作业需要设置此值为longtime

spark.hadoop.odps.cupid.job.capability.duration.hours

  • 流式作业权限文件expired时间,单位小时

spark.hadoop.odps.moye.trackurl.dutation

  • 流式作业jobview expired时间,单位小时


四. 作业诊断

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的存活的时间,存活时间越短的,越有可能是根因所在

      MaxCompute Spark 使用和常见问题

4.2 Spark UI和HistoryServer

  • Spark UI与社区版一致,在logivew的summary模块下找到Spark UI链接:

                       MaxCompute Spark 使用和常见问题

  • 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是社区版本的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')

六. 相关文档


上一篇:解决优盘无法格式化


下一篇:ComputeColStats UDF中 近似算法的介绍(续)