MaxCompute Spark 使用和常见问题

2021-01-12 本文已影响0人阿里云技术

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

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
环境变量配置
参数配置
将$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命令的参数和选项

上传资源：
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的存活的时间，存活时间越短的，越有可能是根因所在

4.2 Spark UI和HistoryServer

Spark UI与社区版一致，在logivew的summary模块下找到Spark UI链接：

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')

原文链接

本文为阿里云原创内容，未经允许不得转载。