摸鱼大数据——Spark基础——Spark On Yarn环境配置和部署

一、背景与问题

在大数据处理场景中，Spark On Yarn 是最常见的运行模式之一。其核心价值在于借助 YARN（Yet Another Resource Negotiator）的资源调度能力，实现 Spark 作业的弹性扩展和资源隔离。然而，实际部署中常遇到以下问题：

1. 资源分配不均：未合理配置内存和核心数导致任务失败
2. 版本兼容性：Spark 与 Hadoop 版本不匹配引发的异常
3. 性能瓶颈：未优化数据分区导致的计算效率低下
4. 安全风险：未配置权限控制导致的数据泄露

本文将深入解析 Spark On Yarn 的底层机制，结合真实开发场景，提供可落地的配置方案。

二、基本原理

1. YARN 架构原理

YARN 采用主从架构，核心组件包括：

• ResourceManager（RM）：全局资源协调者
• NodeManager（NM）：节点资源管理器
• ApplicationMaster（AM）：应用协调器（Spark 通过 AM 管理作业）

YARN 的资源调度流程如下：

1. ApplicationMaster 向 RM 申请资源
2. RM 分配资源并启动 Container
3. NM 启动 Container 执行任务
4. 任务完成后向 AM 汇总结果

2. Spark On YARN 工作流程

Spark On YARN 的关键区别在于：

Spark 作业运行时，Driver 作为 ApplicationMaster 在 YARN 上注册，Executor 作为 Container 运行。

三、环境准备

1. 系统要求

建议使用 CentOS 7+ 系统，安装以下组件：

# 安装 Java 1.8
sudo yum install -y java-1.8.0-openjdk

# 安装 Hadoop 3.3.6
wget https://archive.apache.org/dist/hadoop/core/hadoop-3.3.6/hadoop-3.3.6.tar.gz
tar -zxvf hadoop-3.3.6.tar.gz -C /usr/local
ln -s /usr/local/hadoop-3.3.6 /usr/local/hadoop

# 配置环境变量
export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk
export HADOOP_HOME=/usr/local/hadoop
export PATH=$JAVA_HOME/bin:$HADOOP_HOME/bin:$PATH

2. Spark 安装

下载 Spark 3.3.2（需与 Hadoop 版本匹配）：

wget https://archive.apache.org/dist/spark/spark-3.3.2/spark-3.3.2-bin-hadoop3.3.tgz
tar -zxvf spark-3.3.2-bin-hadoop3.3.tgz -C /usr/local
ln -s /usr/local/spark-3.3.2-bin-hadoop3.3 /usr/local/spark

四、核心实现

1. Spark On YARN 配置

关键配置文件 /usr/local/spark/conf/spark-defaults.conf：

# Spark 集群模式
spark.master                     yarn

# 最大Executor内存（注意：需小于YARN可用内存）
spark.executor.memory           4g

# Executor核心数
spark.executor.cores            2

# 驱动内存
spark.driver.memory             2g

# 本地存储路径
spark.local.dir                 /tmp/spark-local

# 是否启用动态资源分配
spark.dynamicAllocation.enabled true

# 最小/最大Executor数
spark.dynamicAllocation.minExecutors 1
spark.dynamicAllocation.maxExecutors 10

2. 提交作业命令

# 提交Spark作业到YARN
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --class com.example.WordCount \
  --conf spark.executor.memory=4g \
  --conf spark.executor.cores=2 \
  --driver-memory 2g \
  /path/to/your.jar \
  hdfs://namenode/path/input \
  hdfs://namenode/path/output

3. 资源分配优化

// SparkConf 配置示例
val conf = new SparkConf(true)
  .set("spark.yarn.driverMemory", "2g")
  .set("spark.yarn.executorMemory", "4g")
  .set("spark.executor.cores", "2")
  .set("spark.dynamicAllocation.maxExecutors", "10")

五、完整案例

1. 案例描述

处理日志文件统计访问量，要求：

• 读取HDFS日志文件
• 按IP地址统计访问次数
• 输出到HDFS

2. 完整代码示例

// WordCount.scala
object WordCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf(true)
      .set("spark.master", "yarn")
      .set("spark.driver.memory", "2g")
      .set("spark.executor.memory", "4g")
      .set("spark.executor.cores", "2")
      .set("spark.dynamicAllocation.enabled", "true")
      .set("spark.dynamicAllocation.maxExecutors", "10")
    
    val sc = new SparkContext(conf)
    
    val logRDD = sc.textFile("hdfs://namenode/path/to/logs.txt")
    
    val wordCount = logRDD
      .flatMap(line => line.split("\\s+"))
      .map(word => (word, 1))
      .reduceByKey(_ + _)
    
    wordCount.saveAsTextFile("hdfs://namenode/path/output")
    sc.stop()
  }
}

3. 运行流程

1. 提交作业：spark-submit --class WordCount ...
2. YARN 申请资源
3. 启动Driver和Executor
4. 执行计算任务
5. 保存结果

六、源码解析

1. SparkContext 初始化

// SparkContext.scala 源码片段
def initializeContext(): Unit = {
  if (conf.get("spark.master").startsWith("yarn")) {
    // 初始化YARN模式
    yarnConf = new YarnConfiguration(conf)
    yarnConf.set("spark.yarn.driverMemory", "2g")
    yarnConf.set("spark.yarn.executorMemory", "4g")
    yarnConf.set("spark.yarn.executorCores", "2")
    yarnConf.set("spark.dynamicAllocation.maxExecutors", "10")
  }
}

关键点：

• spark.yarn.driverMemory 设置Driver内存
• spark.yarn.executorMemory 设置Executor内存
• spark.yarn.executorCores 设置Executor核心数
• 动态资源分配配置

2. TaskScheduler 初始化

// TaskSchedulerImpl.scala 源码片段
def start(): Unit = {
  if (conf.get("spark.master").startsWith("yarn")) {
    // 初始化YARN TaskScheduler
    val yarnScheduler = new YarnScheduler(this, conf)
    yarnScheduler.start()
  }
}

七、进阶使用

1. 动态资源分配优化

# 启用动态资源分配
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.maxExecutors=10
spark.dynamicAllocation.minExecutors=1
spark.dynamicAllocation.executorAllocationRatio=1.5

2. 内存调优策略

// 内存分配策略配置
spark.executor.memoryOverhead=2g
spark.driver.memoryOverhead=1g
spark.memory.fraction=0.6
spark.memory.storageFraction=0.5

3. 资源隔离策略

# 配置YARN资源隔离
spark.yarn.scheduler.vcorePerExecutor=2
spark.yarn.scheduler.memoryPerExecutor=4g

八、性能与工程实践

1. 性能优化策略

2. 安全风险防范

• 启用 Hadoop Kerberos 认证：

  # 配置kerberos
  spark.kerberos.principal=spark/_hostname@REALM
  spark.kerberos.keytab=/path/to/keytab

• YARN 权限控制：

  # 配置YARN ACL
  yarn.acl_admins=spark_user
  yarn.acl_admins=mapreduce_user

3. 异常处理机制

// 异常处理示例
try {
  val result = spark.sql("SELECT * FROM logs")
  result.show()
} catch {
  case e: Exception =>
    logger.error("作业执行异常", e)
    System.exit(1)
}

九、常见问题与踩坑

1. 常见错误及解决办法

2. 版本兼容性问题

3. 资源分配陷阱

# 错误配置示例
spark.executor.memory=4g  # 导致资源不足
spark.executor.cores=2    # 导致CPU资源浪费

十、最佳实践

1. 推荐配置方案

# 推荐配置参数
spark.master=yarn
spark.executor.memory=4g
spark.executor.cores=2
spark.driver.memory=2g
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.maxExecutors=10
spark.dynamicAllocation.minExecutors=1
spark.dynamicAllocation.executorAllocationRatio=1.5
spark.memory.fraction=0.6
spark.memory.storageFraction=0.5

2. 集群资源规划建议

3. 安全配置建议

• 启用 Kerberos 认证
• 配置 YARN ACL 权限
• 使用加密传输（SSL/TLS）
• 配置审计日志

十一、总结

Spark On Yarn 是大数据处理的首选模式，其核心优势在于：

• 灵活的资源调度能力
• 与 Hadoop 集群的深度集成
• 简化集群管理流程

但需要注意：

• 避免在小规模测试环境使用（资源浪费）
• 需要严格配置资源参数
• 需要关注安全配置

在实际项目中，建议：

• 对复杂计算使用动态资源分配
• 对数据倾斜使用 salting 策略
• 对内存敏感任务进行精细调优
• 对安全要求高的场景启用 Kerberos 认证

通过合理配置和优化，Spark On Yarn 可以实现高效的分布式计算，是现代大数据处理的基石。