CentOS下的分布式内存计算Spark环境部署

一、Spark 核心架构与应用场景

1.1 分布式计算引擎的核心优势

Spark 是基于内存的分布式计算框架，相比 MapReduce 具有以下核心优势：

• 内存计算：数据可常驻内存，迭代计算性能提升 10-100 倍（文档段落：3-790）。
• 多范式支持：同时支持批处理（Spark SQL）、流处理（Spark Streaming）、图计算（GraphX）。
• 生态统一：无缝集成 Hadoop 生态，支持从 HDFS、HBase 等数据源直接读取数据。
• 高吞吐量：单机可处理百万级数据记录，集群可扩展至数千节点。

典型应用场景：

• 实时数据分析（如电商用户行为分析）。
• 大规模机器学习（基于 MLlib 库的模型训练）。
• 流式数据处理（配合 Kafka 实现实时日志分析）。

二、集群环境规划与前置准备

2.1 集群节点规划（3 节点方案）

节点名称	IP 地址	角色分配	内存配置	数据目录
spark-node1	192.168.88.130	Master、Worker	16GB	/data/spark/data
spark-node2	192.168.88.131	Worker	16GB	/data/spark/data
spark-node3	192.168.88.132	Worker	16GB	/data/spark/data

2.2 前置依赖安装（所有节点）

1. JDK 环境（需 1.8+，文档段落：3-248）：

   bash

   yum install -y java-1.8.0-openjdk-devel
   java -version # 验证版本
   

2. Hadoop 集群（已部署 HDFS 和 YARN，文档段落：3-633）：
   确保 HDFS 服务正常，Spark 将使用 HDFS 作为分布式存储。
3. SSH 免密登录（文档段落：3-523）：

   bash

   ssh-keygen -t rsa -b 4096 -N ""
   ssh-copy-id spark-node2 && ssh-copy-id spark-node3
   

三、Spark 单机安装与配置

3.1 下载与解压安装包

bash

# 下载Spark 2.4.5（文档段落：3-796）
wget https://archive.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz# 解压到指定目录
tar -zxvf spark-2.4.5-bin-hadoop2.7.tgz -C /export/server/
ln -s /export/server/spark-2.4.5-bin-hadoop2.7 /export/server/spark # 创建软链接

3.2 核心配置文件修改

3.2.1 spark-env.sh（文档段落：3-800）

bash

vim /export/server/spark/conf/spark-env.sh
# 添加以下内容
export JAVA_HOME=/export/server/jdk
export HADOOP_CONF_DIR=/export/server/hadoop/etc/hadoop
export YARN_CONF_DIR=/export/server/hadoop/etc/hadoop
export SPARK_MASTER_HOST=spark-node1
export SPARK_MASTER_PORT=7077
export SPARK_MASTER_WEBUI_PORT=8080
export SPARK_WORKER_CORES=4 # 每个Worker使用4核
export SPARK_WORKER_MEMORY=8g # 每个Worker使用8GB内存

3.2.2 slaves（文档段落：3-803）

bash

vim /export/server/spark/conf/slaves
# 添加以下内容（每行一个节点）
spark-node1
spark-node2
spark-node3

3.2.3 spark-defaults.conf（可选优化）

bash

cp /export/server/spark/conf/spark-defaults.conf.template /export/server/spark/conf/spark-defaults.conf
vim /export/server/spark/conf/spark-defaults.conf
# 添加以下优化配置
spark.executor.memory 4g
spark.driver.memory 2g
spark.sql.shuffle.partitions 24
spark.default.parallelism 24

四、集群化部署：节点间配置同步

4.1 分发安装目录到其他节点

bash

# 在spark-node1执行，复制到node2/node3
scp -r /export/server/spark spark-node2:/export/server/
scp -r /export/server/spark spark-node3:/export/server/

4.2 配置文件一致性验证

检查所有节点的spark-env.sh和slaves文件内容一致，确保：

• SPARK_MASTER_HOST指向正确的 Master 节点。
• slaves文件包含所有 Worker 节点主机名。

五、集群启动与状态验证

5.1 启动 Spark 集群

5.1.1 单节点启动（spark-node1 执行）

bash

# 启动Master和Worker
/export/server/spark/sbin/start-all.sh

5.1.2 后台启动（生产环境推荐）

bash

nohup /export/server/spark/sbin/start-all.sh &
tail -f /export/server/spark/logs/spark-root-master-spark-node1.log # 查看启动日志

5.2 验证集群状态

5.2.1 进程检查（所有节点执行）

bash

jps | grep -E "Master|Worker"
# spark-node1应显示Master和Worker进程
# spark-node2/node3应显示Worker进程

5.2.2 网页管理界面

• Master 状态：访问http://spark-node1:8080，查看集群概述、Worker 列表、应用程序运行情况（文档段落：3-810）。
• Worker 状态：在管理界面中点击节点名称，查看 CPU、内存、磁盘使用情况。

5.2.3 命令行验证（文档段落：3-812）

bash

# 进入Spark Shell
/export/server/spark/bin/spark-shell --master spark://spark-node1:7077# 执行单词计数示例
val textFile = sc.textFile("hdfs://spark-node1:8020/README.txt")
val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)
wordCounts.collect()

六、核心功能测试与性能调优

6.1 数据处理测试

6.1.1 批量数据处理

使用 Spark SQL 处理 HDFS 上的 Parquet 文件：

sql

-- 在Spark SQL中执行
CREATE TABLE users USING parquet OPTIONS (path "hdfs://spark-node1:8020/users.parquet")
SELECT count(*) FROM users WHERE age > 30

6.1.2 流式数据处理

对接 Kafka 进行实时数据处理（需提前配置 Kafka 主题）：

java

// 示例代码：Spark Streaming对接Kafka
val kafkaParams = Map[String, Object]("bootstrap.servers" -> "kafka-node1:9092,kafka-node2:9092","key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> "test-group","auto.offset.reset" -> "latest","enable.auto.commit" -> (false: java.lang.Boolean)
)
val topics = Array("test-topic")
val stream = KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
)
stream.map(_._2).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()

6.2 性能优化策略

6.2.1 资源分配优化

• Executor 配置：根据任务类型调整 Executor 数量与内存：

  bash

  # 提交任务时指定资源
  /export/server/spark/bin/spark-submit \
  --master spark://spark-node1:7077 \
  --num-executors 10 \
  --executor-memory 4g \
  --executor-cores 2 \
  --class com.example.MyApp \
  my-app.jar
  

• 动态资源分配：在spark-defaults.conf中启用：

  properties

  spark.dynamicAllocation.enabled true
  spark.dynamicAllocation.minExecutors 5
  spark.dynamicAllocation.maxExecutors 50
  

6.2.2 内存优化

• 调整存储与执行内存比例：

  properties

  spark.memory.storageFraction 0.5  # 存储占比50%
  spark.memory.fraction 0.8        # 总内存使用比例80%
  

• 启用堆外内存：

  properties

  spark.memory.offHeap.enabled true
  spark.memory.offHeap.size 2g
  

6.2.3 任务调度优化

• 广播大变量：

  java

  val broadcastVar = sc.broadcast(loadLargeData())
  

• 合并小任务：

  properties

  spark.sql.autoBroadcastJoinThreshold 104857600  # 100MB以下自动广播
  spark.default.parallelism 24 * num-executors
  

七、常见故障排查与解决方案

7.1 集群无法启动

可能原因：

• SSH 免密失败：检查节点间 SSH 连接是否正常，~/.ssh/authorized_keys是否包含所有节点公钥（文档段落：3-523）。
• 端口冲突：确保 Master 端口（7077）、WebUI 端口（8080）未被占用。
• 配置文件错误：检查spark-env.sh中HADOOP_CONF_DIR是否指向正确的 Hadoop 配置目录。

解决方法：

bash

# 示例：修复SSH免密问题
ssh spark-node2 "echo 'hello' > /tmp/test" # 验证连接
netstat -anp | grep 7077 # 检查端口占用

7.2 任务执行缓慢

可能原因：

• 资源分配不足：Executor 数量或内存设置过低，导致任务排队。
• 数据倾斜：某个分区数据量过大，形成任务热点。
• Shuffle 操作频繁：过多的 Shuffle 操作导致磁盘 IO 瓶颈。

解决方法：

• 增加 Executor 数量与内存，调整--num-executors和--executor-memory参数。
• 对倾斜数据进行重分区：

  java

  df.repartition(24).write.parquet("hdfs://path")
  

• 优化 SQL 查询，减少不必要的 Shuffle：

  sql

  SELECT * FROM table WHERE condition ORDER BY id LIMIT 100
  

7.3 Worker 节点掉线

可能原因：

• 内存溢出：Worker 内存不足导致 JVM 崩溃。
• 网络分区：节点间网络延迟过高或断开。
• 磁盘故障：数据目录所在磁盘损坏，导致任务失败。

解决方法：

• 增加spark-worker-memory配置，降低单个任务内存占用。
• 检查网络连接，确保节点间延迟 < 1ms，带宽≥1Gbps。
• 更换故障磁盘，重启 Worker 进程：

  bash

  /export/server/spark/sbin/stop-worker.sh spark-node2:8081
  # 修复磁盘后重新启动
  /export/server/spark/sbin/start-worker.sh spark://spark-node1:7077
  

八、生产环境最佳实践

8.1 高可用性配置

8.1.1 多 Master 部署（HA 模式）

通过 Zookeeper 实现 Master 自动故障转移（文档段落：3-793）：

bash

# 修改spark-env.sh
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.zk.url=zk-node1:2181,zk-node2:2181,zk-node3:2181 -Dspark.zk.dir=/spark"# 启动多个Master
/export/server/spark/sbin/start-master.sh --host spark-node1
/export/server/spark/sbin/start-master.sh --host spark-node2

8.1.2 数据持久化

• Checkpoint 机制：对长时间运行的流作业设置 Checkpoint：

  java

  streamingContext.checkpoint("hdfs://spark-node1:8020/checkpoints")
  

• 作业历史服务器：配置 HistoryServer 记录作业历史：

  bash

  # 修改spark-site.xml
  <property><name>spark.history.fs.logDirectory</name><value>hdfs://spark-node1:8020/spark-logs</value>
  </property>
  /export/server/spark/sbin/start-history-server.sh
  

8.2 安全与资源管理

8.2.1 启用 Kerberos 认证

bash

# 配置spark-env.sh
export SPARK_KERBEROS_PRINCIPAL="[e-mail]"
export SPARK_KERBEROS_KEYTAB="/etc/kerberos/keytabs/spark.keytab"# 提交任务时指定认证
kinit -kt spark.keytab [e-mail]
/export/server/spark/bin/spark-submit --master yarn --deploy-mode cluster ...

8.2.2 资源队列管理

通过 YARN 队列管理不同应用资源（需提前配置 YARN 队列）：

bash

/export/server/spark/bin/spark-submit \
--master yarn --deploy-mode cluster \
--queue production \
--resources memory=8g,vcores=4 \
--class com.example.MyApp my-app.jar

九、总结：Spark 集群部署核心流程

通过以上步骤，可构建一个高可用、高性能的 Spark 分布式计算集群，支撑大规模数据处理与分析任务。生产环境中需结合业务场景优化资源分配、内存使用及任务调度策略，并利用 Spark 生态工具（如 Spark SQL、MLlib）提升开发效率。参考官方文档（Spark Documentation）可进一步学习流处理、机器学习等高级特性及性能调优技巧。