美文网首页技术_工具
MaxCompute Spark 使用和常见问题

MaxCompute Spark 使用和常见问题

作者: 阿里云技术 | 来源:发表于2021-01-12 10:30 被阅读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

spark.hadoop.odps.runtime.end.point

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

原文链接

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

相关文章

网友评论

    本文标题：MaxCompute Spark 使用和常见问题

    本文链接：https://www.haomeiwen.com/subject/zfysaktx.html

    延伸阅读

    深度阅读

    • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

    栏目导航

    热点阅读

    技术_工具

    关于我们|服务条款|联系我们|MaxCompute Spark 使用和常见问题|投稿指南|网站地图|RSS订阅|排版工具|手机版

    提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

    Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

    备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

    本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

    所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！