前言：从开发到上线的鸿沟

很多人觉得，代码在本地跑通了，上线就是分分钟的事。

然后他们遇到了这些问题：

"为什么本地跑得好好的，集群上就OOM？"
"为什么提交任务要等30分钟才开始跑？"
"为什么昨天还正常，今天突然就挂了？"
"日志在哪？我怎么看不到任何错误信息？"

开发环境 vs 生产环境，中间隔着的不是一条河，是太平洋。

本篇是Spark系列的第五篇，我们将聊聊如何把Spark部署到生产环境，以及如何运维一个稳定的Spark集群。

一、Spark部署模式

1.1 部署模式对比

1.2 详细对比表

1.3 如何选择？

二、Spark on YARN详解

2.1 YARN架构回顾

2.2 Client vs Cluster模式

关键区别：

2.3 spark-submit命令详解

# 基本格式
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  [配置选项] \
  <application-jar> \
  [application-arguments]

# 完整示例
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name "MySparkApp" \
  --driver-memory 4g \
  --driver-cores 2 \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 10 \
  --queue production \
  --conf spark.sql.shuffle.partitions=200 \
  --conf spark.default.parallelism=200 \
  --conf spark.yarn.maxAppAttempts=2 \
  --conf spark.yarn.am.memory=2g \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.shuffle.service.enabled=true \
  --jars /path/to/dependency1.jar,/path/to/dependency2.jar \
  --files /path/to/config.properties \
  --class com.example.MyApp \
  /path/to/my-app.jar \
  arg1 arg2

2.4 常用参数说明

2.5 动态资源分配

// spark-defaults.conf 或 spark-submit参数

// 启用动态资源分配
spark.dynamicAllocation.enabled=true

// 启用External Shuffle Service（必须）
spark.shuffle.service.enabled=true

// 初始Executor数量
spark.dynamicAllocation.initialExecutors=5

// 最小Executor数量
spark.dynamicAllocation.minExecutors=2

// 最大Executor数量
spark.dynamicAllocation.maxExecutors=100

// 空闲多久后释放Executor
spark.dynamicAllocation.executorIdleTimeout=60s

// 有缓存数据的Executor空闲多久后释放
spark.dynamicAllocation.cachedExecutorIdleTimeout=300s

// 积压Task多久后增加Executor
spark.dynamicAllocation.schedulerBacklogTimeout=1s

三、Spark on Kubernetes

3.1 为什么上K8s？

3.2 部署方式

3.3 spark-submit方式

# 提交到K8s
spark-submit \
  --master k8s://https://<k8s-apiserver>:443 \
  --deploy-mode cluster \
  --name spark-pi \
  --conf spark.kubernetes.container.image=spark:3.5.0 \
  --conf spark.kubernetes.namespace=spark \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
  --conf spark.executor.instances=3 \
  --conf spark.executor.memory=4g \
  --conf spark.executor.cores=2 \
  --conf spark.driver.memory=2g \
  --conf spark.driver.cores=1 \
  --conf spark.kubernetes.driver.pod.name=spark-pi-driver \
  local:///opt/spark/examples/jars/spark-examples_2.12-3.5.0.jar \
  1000

3.4 Spark Operator方式

# spark-application.yaml
apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pi
  namespace: spark
spec:
  type: Scala
  mode: cluster
  image: spark:3.5.0
  imagePullPolicy: Always
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.0.jar
  arguments:
    - "1000"
  sparkVersion: "3.5.0"
  restartPolicy:
    type: Never
  driver:
    cores: 1
    memory: "2g"
    serviceAccount: spark
    labels:
      version: "3.5.0"
  executor:
    cores: 2
    instances: 3
    memory: "4g"
    labels:
      version: "3.5.0"
  dynamicAllocation:
    enabled: true
    initialExecutors: 2
    minExecutors: 1
    maxExecutors: 10

# 提交应用
kubectl apply -f spark-application.yaml

# 查看状态
kubectl get sparkapplication -n spark

# 查看日志
kubectl logs spark-pi-driver -n spark

3.5 K8s配置参数

四、配置管理

4.1 配置优先级

4.2 spark-defaults.conf

# /etc/spark/conf/spark-defaults.conf

# 基本配置
spark.master                     yarn
spark.submit.deployMode          cluster
spark.driver.memory              4g
spark.executor.memory            8g
spark.executor.cores             4

# SQL配置
spark.sql.shuffle.partitions     200
spark.sql.adaptive.enabled       true
spark.sql.adaptive.coalescePartitions.enabled true

# 动态资源分配
spark.dynamicAllocation.enabled  true
spark.shuffle.service.enabled    true
spark.dynamicAllocation.minExecutors 2
spark.dynamicAllocation.maxExecutors 100

# 序列化
spark.serializer                 org.apache.spark.serializer.KryoSerializer
spark.kryoserializer.buffer.max  256m

# 日志
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs:///spark-history
spark.history.fs.logDirectory    hdfs:///spark-history

# 推测执行
spark.speculation                true
spark.speculation.interval       100ms
spark.speculation.multiplier     1.5
spark.speculation.quantile       0.75

# YARN配置
spark.yarn.queue                 production
spark.yarn.maxAppAttempts        2

4.3 spark-env.sh

#!/bin/bash
# /etc/spark/conf/spark-env.sh

# Java配置
export JAVA_HOME=/usr/lib/jvm/java-11
export SPARK_JAVA_OPTS="-Xms1g -Xmx1g"

# Hadoop配置
export HADOOP_HOME=/opt/hadoop
export HADOOP_CONF_DIR=/etc/hadoop/conf
export YARN_CONF_DIR=/etc/hadoop/conf

# Spark配置
export SPARK_HOME=/opt/spark
export SPARK_CONF_DIR=/etc/spark/conf
export SPARK_LOG_DIR=/var/log/spark
export SPARK_PID_DIR=/var/run/spark

# Standalone模式配置
export SPARK_MASTER_HOST=spark-master
export SPARK_MASTER_PORT=7077
export SPARK_MASTER_WEBUI_PORT=8080
export SPARK_WORKER_MEMORY=32g
export SPARK_WORKER_CORES=8

# History Server
export SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs:///spark-history"

4.4 log4j2.properties

# /etc/spark/conf/log4j2.properties

# Root logger
rootLogger.level = WARN
rootLogger.appenderRef.console.ref = console

# Console appender
appender.console.type = Console
appender.console.name = console
appender.console.target = SYSTEM_ERR
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n

# Spark日志级别
logger.spark.name = org.apache.spark
logger.spark.level = WARN

logger.hadoop.name = org.apache.hadoop
logger.hadoop.level = WARN

# 自定义应用日志
logger.myapp.name = com.example.myapp
logger.myapp.level = INFO

五、History Server

5.1 为什么需要History Server？

5.2 配置History Server

# 1. 创建日志目录
hdfs dfs -mkdir -p /spark-history
hdfs dfs -chmod 777 /spark-history

# 2. 配置spark-defaults.conf
# 启用事件日志
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs:///spark-history
spark.eventLog.compress          true

# History Server配置
spark.history.fs.logDirectory    hdfs:///spark-history
spark.history.fs.cleaner.enabled true
spark.history.fs.cleaner.interval 1d
spark.history.fs.cleaner.maxAge  7d
spark.history.retainedApplications 50

# 3. 启动History Server
$SPARK_HOME/sbin/start-history-server.sh

# 4. 访问Web UI
# http://history-server:18080

5.3 History Server架构

六、日志管理

6.1 日志类型

6.2 查看YARN日志

# 查看应用日志（全部）
yarn logs -applicationId application_1234567890_0001

# 查看特定Container日志
yarn logs -applicationId application_1234567890_0001 \
  -containerId container_1234567890_0001_01_000001

# 查看Driver日志（AM日志）
yarn logs -applicationId application_1234567890_0001 \
  -containerId container_1234567890_0001_01_000001 \
  -log_files stdout,stderr

# 实时查看日志（运行中的应用）
yarn logs -applicationId application_1234567890_0001 -follow

# 下载日志到本地
yarn logs -applicationId application_1234567890_0001 > app.log

6.3 日志聚合配置

<!-- yarn-site.xml -->

<!-- 启用日志聚合 -->
<property>
  <name>yarn.log-aggregation-enable</name>
  <value>true</value>
</property>

<!-- 聚合日志保留时间 -->
<property>
  <name>yarn.log-aggregation.retain-seconds</name>
  <value>604800</value> <!-- 7天 -->
</property>

<!-- 聚合日志存储路径 -->
<property>
  <name>yarn.nodemanager.remote-app-log-dir</name>
  <value>/var/log/hadoop-yarn/apps</value>
</property>

6.4 自定义日志

// 在代码中使用日志
import org.apache.spark.internal.Logging

class MyApp extends Logging {
  def run(): Unit = {
    logInfo("Application started")
    logDebug("Debug information")
    logWarning("This is a warning")
    logError("This is an error")
  }
}

// 或使用标准SLF4J
import org.slf4j.LoggerFactory

object MyApp {
  private val logger = LoggerFactory.getLogger(this.getClass)
  
  def main(args: Array[String]): Unit = {
    logger.info("Starting application")
    // ...
  }
}

七、监控体系

7.1 监控架构

7.2 Prometheus + Grafana

# spark-defaults.conf

# 启用Prometheus监控
spark.metrics.conf.*.sink.prometheusServlet.class=org.apache.spark.metrics.sink.PrometheusServlet
spark.metrics.conf.*.sink.prometheusServlet.path=/metrics/prometheus
spark.metrics.conf.master.sink.prometheusServlet.path=/metrics/master/prometheus
spark.metrics.conf.applications.sink.prometheusServlet.path=/metrics/applications/prometheus

# 或使用Pushgateway
spark.metrics.conf.*.sink.graphite.class=org.apache.spark.metrics.sink.GraphiteSink
spark.metrics.conf.*.sink.graphite.host=prometheus-pushgateway
spark.metrics.conf.*.sink.graphite.port=9091

# prometheus.yml
scrape_configs:
  - job_name: 'spark'
    static_configs:
      - targets: ['spark-master:4040']
    metrics_path: /metrics/prometheus
    
  - job_name: 'spark-history'
    static_configs:
      - targets: ['history-server:18080']
    metrics_path: /metrics/prometheus

7.3 关键监控指标

7.4 Grafana Dashboard示例

{
  "dashboard": {
    "title": "Spark Monitoring",
    "panels": [
      {
        "title": "Executor Count",
        "type": "stat",
        "targets": [{
          "expr": "spark_executor_count"
        }]
      },
      {
        "title": "Memory Usage",
        "type": "gauge",
        "targets": [{
          "expr": "spark_executor_memory_used / spark_executor_memory_total * 100"
        }]
      },
      {
        "title": "Active Tasks",
        "type": "graph",
        "targets": [{
          "expr": "spark_scheduler_active_tasks"
        }]
      },
      {
        "title": "GC Time",
        "type": "graph",
        "targets": [{
          "expr": "rate(spark_jvm_gc_time[5m])"
        }]
      }
    ]
  }
}

八、故障排查

8.1 常见问题排查流程

8.2 OOM问题排查

# 1. 确定是哪里OOM
# Driver OOM
java.lang.OutOfMemoryError: Java heap space
  at ... (Driver堆栈)

# Executor OOM
ExecutorLostFailure (executor X exited caused by one of the running tasks)
Reason: Container killed by YARN for exceeding memory limits

# 2. 查看内存使用
# Spark UI -> Executors -> Memory Usage

# 3. 解决方案
# Driver OOM
--driver-memory 8g
--conf spark.driver.maxResultSize=4g

# Executor OOM
--executor-memory 16g
--conf spark.memory.fraction=0.8
--conf spark.sql.shuffle.partitions=500

8.3 数据倾斜排查

// 1. 查看分区数据分布
df.groupBy(spark_partition_id()).count().show()

// 2. 查看Key分布
df.groupBy("key_column").count().orderBy($"count".desc).show(20)

// 3. 在Spark UI查看
// Stages -> 点击Stage -> Summary Metrics
// 查看各Task的输入数据量差异

8.4 常见错误及解决

九、安全配置

9.1 安全架构

9.2 Kerberos配置

# spark-defaults.conf

# 启用Kerberos
spark.yarn.principal=spark@EXAMPLE.COM
spark.yarn.keytab=/etc/security/keytabs/spark.keytab

# HDFS安全配置
spark.hadoop.dfs.namenode.kerberos.principal=hdfs/_HOST@EXAMPLE.COM

# Hive安全配置
spark.hadoop.hive.metastore.kerberos.principal=hive/_HOST@EXAMPLE.COM

9.3 传输加密

# spark-defaults.conf

# 启用RPC加密
spark.authenticate=true
spark.authenticate.secret=your-secret-key

# 启用SSL
spark.ssl.enabled=true
spark.ssl.keyPassword=your-keystore-password
spark.ssl.keyStore=/path/to/keystore.jks
spark.ssl.keyStorePassword=your-keystore-password
spark.ssl.trustStore=/path/to/truststore.jks
spark.ssl.trustStorePassword=your-truststore-password

# 网络加密
spark.network.crypto.enabled=true
spark.network.crypto.keyLength=256

十、最佳实践Checklist

10.1 上线前Checklist

## Spark应用上线Checklist

### 代码检查
- [ ] 代码已通过Code Review
- [ ] 单元测试通过
- [ ] 在测试环境验证通过
- [ ] 没有打印敏感信息

### 资源配置
- [ ] Driver内存合理（2-4g）
- [ ] Executor内存合理（4-8g）
- [ ] Executor核数合理（2-4）
- [ ] 分区数合理（避免过多/过少）
- [ ] 配置了动态资源分配

### 运行配置
- [ ] 使用cluster模式
- [ ] 配置了正确的YARN队列
- [ ] 配置了重试次数
- [ ] 配置了推测执行

### 监控告警
- [ ] 配置了日志
- [ ] 配置了监控
- [ ] 配置了告警
- [ ] 有值班人员

### 容错恢复
- [ ] 配置了Checkpoint（流处理）
- [ ] 有回滚方案
- [ ] 有应急预案

10.2 资源配置建议

10.3 参数调优顺序

十一、写在最后

部署和运维Spark，是一门需要长期积累的技术。

记住几个原则：

1. 监控先行：没有监控的系统是盲人骑瞎马
2. 日志要全：出问题时日志是你唯一的朋友
3. 配置要稳：不要在生产环境随便调参数
4. 预案要有：任何系统都会出问题，关键是出了问题怎么办
5. 文档要写：你能记住的，同事不一定能记住

最后送大家一句话：

线上无小事，每一次变更都要如履薄冰。

本文作者：一个凌晨3点被电话叫醒的运维人

最惨经历：周五下午上线，周六凌晨回滚

下一篇：Spark MLlib实战

附录：面试高频题

1. Spark on YARN的Client和Cluster模式有什么区别？

   Client模式Driver在提交机器，适合调试；Cluster模式Driver在集群中，适合生产。

2. 动态资源分配是什么？怎么配置？

   根据负载自动调整Executor数量。需要启用dynamicAllocation和shuffle.service。

3. 如何查看Spark任务的日志？

   YARN模式用yarn logs命令；Spark UI实时查看；History Server查看历史。

4. Spark任务OOM怎么排查？

   确定是Driver还是Executor OOM，查看Spark UI内存使用，增加内存或优化代码。

5. 如何监控Spark应用？

   Spark Web UI、Prometheus+Grafana、History Server、自定义Metrics。

6. 生产环境部署Spark有哪些注意事项？

   使用cluster模式、配置动态资源、启用日志和监控、配置重试和推测执行、做好安全配置。