跳到主要内容

Spark 集成指南

概述​

Apache Spark 是用于大规模数据处理的统一分析引擎。它提供快速、通用的集群计算平台，支持批处理、流处理、机器学习和图计算。

核心特性​

  • ⚡ 快速 - 基于内存的计算，比 Hadoop MapReduce 快 100 倍
  • 🔄 通用 - SQL 查询、流处理、机器学习、图计算
  • 📦 易用 - 支持 Scala、Java、Python、R 多种语言
  • 🌐 灵活 - 可运行在 Hadoop、Kubernetes、Mesos、Standalone 等多种环境

与 DolphinScheduler 集成​

集成方式​

DolphinScheduler 提供 Spark Submit 任务类型，可以直接提交 Spark 作业。

前置要求​

# 安装 Spark
sudo apt install openjdk-17-jdk

# 下载 Spark
cd /tmp
wget https://archive.apache.org/dist/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz
tar -xzf spark-3.5.0-bin-hadoop3.tgz
sudo mv spark-3.5.0-bin-hadoop3 /opt/spark

# 配置环境变量
export SPARK_HOME=/opt/spark
export PATH=$SPARK_HOME/bin:$PATH

在 DS 中创建 Spark 任务​

方法一：使用 Spark Submit 任务类型​

在 DS Web UI 中选择任务类型为 SPARK：

参数配置：

# Program Name
programName=spark-example

# Main Class
mainClass=com.example.SparkJob

# Deploy Mode
deployMode=cluster

# Master
master=yarn

# Arguments
--conf spark.executor.instances=2 \
--conf spark.executor.memory=2g \
--conf spark.driver.memory=1g \
/opt/spark/jobs/spark-job.jar arg1 arg2

方法二：使用 Shell 任务​

#!/bin/bash

source /opt/dolphinscheduler/conf/dolphinscheduler_env.sh

$SPARK_HOME/bin/spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --class com.example.WordCount \
  --executor-memory 2g \
  --executor-cores 2 \
  --num-executors 10 \
  /opt/spark/jobs/spark-wordcount.jar \
  hdfs://namenode:9000/input \
  hdfs://namenode:9000/output

典型应用场景​

1. ETL 数据处理​

// Spark ETL 示例
import org.apache.spark.sql._

val df = spark.read.parquet("hdfs://namenode:9000/raw_data")

val cleaned = df
  .filter($"age" >= 18)
  .withColumnRenamed("name", "user_name")
  .dropDuplicates()

cleaned.write
  .mode("overwrite")
  .partitionBy("date")
  .parquet("hdfs://namenode:9000/cleaned_data")

DS 任务配置：

spark-submit \
  --master yarn \
  --class com.example.ETLJob \
  /opt/spark/jobs/etl-job.jar \
  hdfs://namenode:9000/raw_data \
  hdfs://namenode:9000/cleaned_data

2. 实时流处理​

// Spark Streaming 示例
import org.apache.spark.streaming._

val ssc = new StreamingContext(sc, Seconds(5))

val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)

wordCounts.print()
ssc.start()
ssc.awaitTermination()

3. 机器学习​

// Spark MLlib 示例
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.RandomForestClassifier

val trainingData = spark.read.parquet("hdfs://namenode:9000/training")

val rf = new RandomForestClassifier()
  .setLabelCol("label")
  .setFeaturesCol("features")
  .setNumTrees(100)

val model = rf.fit(trainingData)

model.save("hdfs://namenode:9000/rf_model")

性能优化​

1. 资源配置​

spark-submit \
  --executor-cores 4 \      # 每个 Executor 的 CPU 核心数
  --executor-memory 8g \    # 每个 Executor 的内存
  --num-executors 20 \      # Executor 数量
  --driver-memory 4g \      # Driver 内存
  --conf spark.default.parallelism=200 \
  --conf spark.sql.shuffle.partitions=200

2. 数据倾斜处理​

// 加盐处理
val saltedDF = df.withColumn("salt", (rand() * 10).cast("int"))
val joinedDF = df1.join(df2, Seq("key", "salt"))

3. 缓存策略​

// 持久化到内存
df.cache()
df.persist(StorageLevel.MEMORY_AND_DISK)

// 清理缓存
df.unpersist()

监控与调优​

Spark UI​

关键指标​

  • Task Duration
  • Shuffle Read/Write
  • GC Time
  • Memory Usage

最佳实践​

1. 开发规范​

  • 使用 DataFrame/Dataset API（优于 RDD）
  • 避免使用 * 选择所有列
  • 合理使用分区和分桶
  • 及时释放不需要的缓存

2. 生产部署​

  • 使用 YARN/K8s 集群模式
  • 配置合理的资源配额
  • 启用动态资源分配
  • 设置任务超时时间

3. 故障恢复​

spark-submit \
  --conf spark.yarn.maxAppAttempts=3 \
  --conf spark.task.maxFailures=10 \
  --conf spark.speculation=true

参考资源​