企业官网建设流程全解析

简介：围绕Spark 3.0.1的入门到精通学习资源，主要面向大数据初学者和希望系统进阶的开发者，以2021贺岁课程为蓝本，覆盖环境搭建、Spark Core、Spark Streaming、Spark SQL、Structured Streaming、综合案例、多语言开发、3.0新特性与性能调优等九大主题，构成一条从基础概念到实战调优的完整学习路径。资源包共244个文件，总大小86.9MB，以png截图、md笔记、scala源码为主，并附带9个zip子包和1个json配置。其中217张截图直观展示安装与运行界面，10份markdown笔记归纳每节重点，7个scala源码文件供动手实践，子zip包则可进一步拆分课程素材，便于按需学习。已有609人学习参考。按1至8天组织，每天配套独立笔记与代码，可配合课程搭建Spark集群，理解RDD与DataFrame等核心抽象，掌握流式处理与结构化查询的开发方式，同时了解3.0版本的自适应查询执行、动态分区裁剪等新特性及常用调优方法。对于零基础入门或准备系统梳理Spark知识的学习者，这套资源既提供阶段化指引，又能通过代码与截图的结合有效降低上手门槛，适合自学及复习巩固。

1. Spark 3.0 大数据入门：这份 8 天笔记为什么值得照着敲

在聊大数据学习路线的时候，很多人第一站就卡在 Spark 上：Hadoop 那套还没吃透，Spark 又抛来 RDD、DataFrame、Streaming 一堆概念，真到集群上跑又全是环境问题。这份课程用的不是老版本，而是官方 2020 年 9 月 8 日发布的 Spark 3.0.1 稳定版，从 Spark 环境搭建一路打到性能调优，8 天的代码和笔记全在一起，算是一条能把大数据入门到精通串完整的路径。适合刚学完 Hadoop、准备啃 Spark 的初学者，也适合想快速接手 Spark 项目但缺一份完整代码参考的开发。整套资源里既有按天拆好的 Spark-day01 到 Spark-day06 笔记，也有独立的 Spark 综合案例、Spark 多语言开发，以及配套的 question_info.json，照着敲比自己瞎试快得多。

2. Spark 环境搭建与 SparkCore：先跑通第一个分布式任务再谈优化

2.1 版本选型：为什么锁 Spark 3.0.1 而不是旧版

拿到这份资源，第一步不是急着写代码，而是把 Spark 3.0.1 的运行环境对齐。这个版本是 Apache 官方 2020 年 9 月发布的 3.0 系稳定版本，编译基础是 Scala 2.12，官方对 Hadoop 的兼容范围覆盖 2.7 和 3.2。JDK 方面，Spark 3.0 要求 8 或 11，我自己习惯选 JDK 8u202 以上版本，因为后续 Spark SQL 和 Hive 元数据交互时，旧版 JDK 偶尔会抛奇怪的反射异常，排查起来非常浪费时间。

# 我习惯的目录规划：多套版本切换时不互相污染 mkdir -p /opt/bigdata/spark /opt/bigdata/hadoop /opt/bigdata/jdk tar -zxvf spark-3.0.1-bin-hadoop2.7.tgz -C /opt/bigdata/spark mv /opt/bigdata/spark/spark-3.0.1-bin-hadoop2.7 /opt/bigdata/spark/spark-3.0.1

这段命令的作用是解压 Spark 安装包，并把长目录名改成短路径。这里有个选型细节容易被忽略：下载时一定选名字里带bin-hadoop2.7或bin-hadoop3.2字样的预编译包，别下src结尾的源码包。源码包拿回来要用 Maven 重新编译，时长半小时起步，而且编译环境不一致时spark-shell直接起不来，属于最典型的入门回收站行为。

环境变量是另一个高频翻车点。Spark 3.0.1 启动时同时读JAVA_HOME和HADOOP_HOME，少配一个就报找不到可执行文件。我一般把配置写进/etc/profile.d/spark.sh里，这样切用户之后还能正常识别。

export JAVA_HOME=/opt/bigdata/jdk/jdk1.8.0_202 export HADOOP_HOME=/opt/bigdata/hadoop/hadoop-3.2.1 export SPARK_HOME=/opt/bigdata/spark/spark-3.0.1 export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$SPARK_HOME/bin

配置完执行spark-shell --version能打出版本号，环境就算通了一半。接下来马上进入运行模式选型，这块在 Spark-day01.md 里有专门对照表，我这里也列了一份：

前三天用local[*]完全够，没必要一上来就搭三台虚拟机。我见过不少人第一天就在配集群，结果环境没弄好，WordCount 都没跑通，信心先被打了一半。

2.2 RDD 核心算子：从 WordCount 理解完整数据流转

Spark-day01 和 Spark-day02 里，WordCount 是反复出现的例子。它看起来简单，却把 RDD 的三个关键特性全带出来了：不落盘、懒加载、分区并行。理解这三件事，后面调优时才有抓手。

from pyspark import SparkContext, SparkConf conf = SparkConf().setAppName("wordcount_demo").setMaster("local[*]") sc = SparkContext(conf=conf) lines = sc.textFile("file:///data/input/words.txt", minPartitions=2) words = lines.flatMap(lambda line: line.split(" ")) pairs = words.map(lambda word: (word, 1)) counts = pairs.reduceByKey(lambda a, b: a + b) counts.saveAsTextFile("file:///data/output/wc_result")

代码逻辑分四层：textFile读取文本并按参数切分分区，flatMap把每行打散成单词，map把单词映射成(word, 1)二元组，reduceByKey先做分区内局部聚合，再做跨分区全局聚合。saveAsTextFile有个强制约束：输出目录必须不存在，否则直接跑FileAlreadyExistsException，这是 Spark 对输出路径的保护机制，踩过一次就不会忘。

minPartitions=2值得单独讲。它是期望的最小分区数，不是绝对保证，实际分区数还看文件存储方式：本地文件系统默认按 32MB 切块，HDFS 按 128MB 切块。从这个小参数能引出调优课里的一个重要原则：分区数太少，单个任务处理数据量过重；分区数太多，调度开销反而拉低吞吐。常见经验值是让每个分区承载约 128MB 数据量。

2.3 DataFrame 转换：Catalyst 优化器带来的思维切换

到 Spark-day02 后半部分，笔记开始引入 DataFrame。这里要做一个思维切换：RDD 关注的是数据如何被分散处理，DataFrame 关注的是表结构上的声明式操作，底层 Catalyst 优化器会自动把filter、join的执行顺序重排到更优路径。

import org.apache.spark.sql.SparkSession val spark = SparkSession.builder() .appName("rdd_to_df") .master("local[*]") .getOrCreate() import spark.implicits._ val rdd = spark.sparkContext.textFile("file:///data/input/sales.txt") val df = rdd.map { line => val fields = line.split(",") (fields(0), fields(1).toDouble) }.toDF("category", "amount") df.filter($"amount" > 100) .groupBy("category") .agg(sum("amount").as("total_amount")) .show()

这段代码有两个隐性知识点。第一次用toDF的人，十有八九会漏掉import spark.implicits._，导致toDF和$"amount"编译报错，这是 Scala 隐式转换机制决定的。第二个坑是 DataFrame 的算子全部返回新对象，filter之后原df不受影响，这和 RDD 的不可变语义一致，但刚接触时总有人觉得应该像 Python 列表那样原地修改。

Day02 笔记里有一段 RDD 和 DataFrame 在执行计划层面的对比。RDD 的filter是直接作用于分区的函数调用，DataFrame 的filter先生成逻辑计划节点，Catalyst 再结合统计信息决定是否做谓词下推。这种优化在 Parquet 列式存储上效果尤其显著，后面的综合案例做数据清洗时会反复用到。从这个角度看，Spark Core 阶段真正要练的不是算子数量，而是理解数据流转的直觉：数据从哪来，经过哪些转换，最后落到哪。

2.4 序列化与依赖：本地跑通不等于集群能跑通

单机用local[*]跑通 WordCount 只是第一步，把任务真正丢到集群上时，碰到的第一个大坑基本是序列化。Spark 3.0.1 在集群模式下，driver 把任务分发给 executor 时，闭包里引用的对象要通过 Java 序列化传输。自定义类不实现Serializable，提交任务时就会报Task not serializable。

class MyFilter extends Serializable { val threshold: Double = 100.0 def isHigh(amount: Double): Boolean = amount > threshold } val myFilter = new MyFilter() df.filter(row => myFilter.isHigh(row.getAs[Double]("amount"))).count()

解决方案是让闭包中引用的类实现Serializable，或者把常量提取成方法内局部变量，避免闭包捕获整个对象。Spark-day02.md 对这个问题有单独标注，因为综合案例里自定义 UDF 时还会再踩一次。记住一个判断标准：凡是 driver 端定义、executor 端使用的对象，都必须过序列化这一关。

3. Spark SQL 与流处理：从 DataFrame API 到 Structured Streaming 的实战配置

3.1 Spark SQL 读写通道：Parquet、JDBC 与 Hive 表的打通

Spark SQL 这部分课程内容，核心不是 SQL 语法本身，而是读写通道怎么打通。Spark 3.0.1 的 SQL 可以读 HDFS 上的 Parquet、JSON、CSV，也能通过 JDBC 连 MySQL，还能挂上 Hive 元数据服务。实操里我优先选 Parquet，原因是列式存储加压缩，配合 Catalyst 的谓词下推，全表扫描的量能砍掉一大截。

# PySpark 写法：读取 Parquet 后注册临时表，再跑纯 SQL from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("spark_sql_demo") \ .config("spark.sql.adaptive.enabled", "true") \ .getOrCreate() df = spark.read.parquet("hdfs:///data/input/sales.parquet") df.createOrReplaceTempView("sales") result = spark.sql(""" SELECT category, count(*) AS order_cnt, round(sum(amount), 2) AS total_amount FROM sales WHERE dt = '2024-01-15' GROUP BY category ORDER BY total_amount DESC """) result.show()

这个例子里，createOrReplaceTempView是关键动作，它把 DataFrame 注册成临时表，之后就能用纯 SQL 操作。Spark 3.0 里临时表默认只在当前 SparkSession 内可见，退出就消失，想做跨会话共享得用saveAsTable落成持久表。

CSV 读数的坑也值得单独说。CSV 没有 Schema 信息，Spark 默认会把所有列当字符串读，后面做聚合时才发现类型不对。

# 指定 schema 避免类型推错 from pyspark.sql.types import StructType, StructField, StringType, DoubleType schema = StructType([ StructField("category", StringType(), True), StructField("amount", DoubleType(), True), StructField("dt", StringType(), True) ]) df = spark.read \ .option("header", "true") \ .schema(schema) \ .csv("hdfs:///data/input/sales.csv")

这里header选项控制首行是否作为列名，schema显式声明列类型。不写 schema 临时能跑，但数据量一大或者源文件里混入空值，类型推错的问题会延迟到聚合阶段才爆出来，那时候排查成本就高了。日志和数据质量验证应该在数据接入的那一刻就做，而不是等结果对不上再去翻数据。

3.2 Spark Streaming 与 Structured Streaming：两代流处理的选型

Spark-day03 和 day04 开始进入流处理，这里要分清两个概念：Spark Streaming 是早期的 DStream 模型，按微批处理，API 和 RDD 一脉相承；Structured Streaming 是 Spark 2.0 之后引入的基于 DataFrame 的流处理模型，到 3.0 已经进入成熟期，支持mapGroupsWithState、flatMapGroupsWithState这些有状态算子。

选型上，新项目直接学 Structured Streaming，除非要维护老代码。理由很现实：Structured Streaming 的 API 和 Spark SQL 几乎一致，能复用批处理里的思维，而且 3.0 的 Streaming 和 SQL 的执行引擎已经统一到同一套 Catalyst 优化路径上。

# Structured Streaming 读取 Kafka 的典型写法 from pyspark.sql import SparkSession from pyspark.sql.functions import window, col spark = SparkSession.builder \ .appName("streaming_kafka") \ .config("spark.sql.shuffle.partitions", "8") \ .getOrCreate() df = spark.readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "localhost:9092") \ .option("subscribe", "user_actions") \ .option("startingOffsets", "latest") \ .load() parsed = df.selectExpr("CAST(value AS STRING) as json") \ .selectExpr("from_json(json, 'user_id STRING, action STRING, ts TIMESTAMP') as data") \ .select("data.*") windowed = parsed \ .withWatermark("ts", "10 minutes") \ .groupBy(window("ts", "5 minutes"), "action") \ .count() query = windowed.writeStream \ .outputMode("append") \ .format("console") \ .trigger(processingTime="5 seconds") \ .start() query.awaitTermination()

这段代码里值得展开的点：withWatermark设置水位线，用来容忍乱序数据；window("ts", "5 minutes")定义滚动窗口；trigger(processingTime="5 seconds")控制微批间隔。startingOffsets设为latest表示只消费启动后的增量数据，调试阶段这样最快；生产环境一般用earliest配合 checkpoint 做断点续跑。

3.3 流式任务的 checkpoint 与语义：至少一次与精确一次的差别

流处理项目里，checkpoint 不是可选配置，是必选。Structured Streaming 把 Offset 和状态数据都写进 checkpoint 目录，用来做故障恢复。不配 checkpoint，任务只要重启，offset 就找不回来了。

query = windowed.writeStream \ .format("parquet") \ .option("path", "hdfs:///data/output/stream_result") \ .option("checkpointLocation", "hdfs:///data/checkpoint/stream_result") \ .outputMode("append") \ .trigger(processingTime="10 seconds") \ .start()

checkpointLocation必须绑定一个不和其他任务共用的目录。如果两个流任务共用一个 checkpoint 路径，重启时会因为目录里已经有 driver 元数据而直接拒绝启动。这个坑我在联调时遇到过，现象是日志一直刷A checkpoint directory must not be used by multiple applications，原因就是复制粘贴时没改路径。

处理语义方面，Spark 3.0.1 的 Structured Streaming 默认提供至少一次语义，要精确一次需要额外配合 Kafka 的幂等写出和事务日志。对入门阶段来说，先保证至少一次，再把重复数据在聚合层做去重，比一上来就追求精确一次更实际。

4. Spark 性能调优与常见问题排查：从执行计划到 5 个真实现场

4.1 用 Web UI 和物理执行计划定位瓶颈

课程后半部分反复强调：调优之前先看 UI。Spark 3.0.1 的 Web UI 默认跑在 4040 端口，包含 Jobs、Stages、Storage、Executors 四个关键页签。看到一个耗时异常长的 Stage，点进去看每个 Task 的处理时间分布，基本上能判断是数据倾斜还是资源不足。

-- 查看物理执行计划，SQL 里加 EXPLAIN EXPLAIN COST SELECT category, sum(amount) FROM sales WHERE dt = '2024-01-15' GROUP BY category;

EXPLAIN COST会输出优化后的逻辑计划和带行数估算的物理计划。如果发现Exchange节点两侧数据量差距过大，基本可以确定是 join 键分布不均导致的倾斜。另一个常用手段是把计划打到日志里，用df.explain(true)在 DataFrame API 里直接看完整计划。

4.2 资源参数配置：executor 数量、内存与并行度怎么定

Spark-day07 性能调优笔记里最实用的内容，是一套资源参数的初始值设定方法。很多新手上来就调spark.executor.memory，但忽略了 executor 数量和并行度之间的配合，导致内存够用但 CPU 利用率极低。

# 常见 submit 参数模板 spark-submit \ --master yarn \ --deploy-mode cluster \ --num-executors 20 \ --executor-cores 4 \ --executor-memory 8g \ --driver-memory 4g \ --conf spark.sql.shuffle.partitions=200 \ --class com.example.MainJob /data/app/spark-demo.jar

关键参数的含义我按经验列一张表：

一个容易踩的坑是 executor 内存只算堆内，没算堆外。spark.executor.memoryOverhead需要单独留，尤其在跑 Python UDF 和加解密类任务时，堆外内存不足会抛Container killed by YARN for exceeding memory limits。出现这个错误，优先加memoryOverhead，而不是继续加堆内内存。

4.3 五个常见问题的现象、原因、解决

问题一：java.lang.OutOfMemoryError: GC overhead limit exceeded

现象：任务跑到 shuffle 阶段频繁 Full GC，然后直接 OOM。

原因：executor 内存中spark.memory.storage和spark.memory.execution存在动态抢占，默认各占 50%，但如果 RDD cache 数据过量，又同时跑大聚合，内存就被挤爆了。

解决：先检查代码里是否有过量的cache()或persist()，及时unpersist()；再调spark.memory.fraction从默认 0.6 适当降到 0.5，给 JVM 留更多余量；最后把 executor 内存加大一到两档。

问题二：java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE

现象：groupByKey或reduceByKey在单个分区内累积的数据超过 2GB。

原因：数据倾斜导致某个 key 的数据量巨大，单分区无法容纳。

解决：把这个算子改成两阶段聚合，先加随机盐键打散，再聚合去盐。课程综合案例里有个按用户维度聚合的场景就是这么处理的。

问题三：Caused by: java.io.FileNotFoundException: File does not exist

现象：任务重启后读 HDFS 上的临时文件失败。

原因：Spark 在 shuffle 过程中生成的临时文件被定期清理，任务重试时旧文件句柄已经失效。这个问题在开启动态资源分配时尤其常见。

解决：关闭动态资源分配spark.dynamicAllocation.enabled=false，或者把spark.shuffle.service.enabled=true开启，让 shuffle 文件由外部服务托管。

问题四：org.apache.spark.SparkException: Task not serializable

现象：提交任务时直接报错，栈指向自定义过滤类。

原因：闭包捕获了不可序列化的对象。我在 2.4 节踩过一模一样的坑。

解决：类实现Serializable，或把成员变量提取成局部变量传入函数。

问题五：argument type mismatch或Column 'xxx' does not exist

现象：DataFrame 的列名和查询不一致，运行时报错。

原因：CSV 或 JSON 数据源列名大小写不一致，或者读入时列名被 Spark 改了，比如带逗号的列名会被加上反引号。

解决：统一在数据接入层做列名清洗，用withColumnRenamed映射成规范命名，再用df.printSchema()验证。这个习惯能省掉后续 SQL 阶段大量莫名其妙的报错。

5. Spark 综合案例与多语言开发：把 8 天内容串成一个可复现项目

5.1 综合案例拆解：从数据清洗到指标输出的完整链路

Spark 综合案例.md 是整个课程里最值得反复读的文件。它不只是写几段代码，而是把用户行为数据从原始日志到最终报表的完整处理链路串了起来。链路的每一段，正好对应前面 8 天的某个知识点：数据接入对应 Spark SQL，实时清理对应 Structured Streaming，离线统计对应 SparkCore，最终输出又落回 Spark SQL 写 Hive 或 Parquet。

实际去做这类案例时，我一般按四步拆：先定义需求，即最终要输出哪些表和指标；再定数据源，分清离线文件、Kafka 流和数据库表；然后画数据流，标出每个转换节点的输入输出；最后才写代码。整个链路里最容易出问题的步骤是数据清洗，空值处理策略不统一会导致结果数据凭空多出很多脏行。

# 综合案例中常见的数据清洗片段 from pyspark.sql.functions import col, when, isnan df_clean = df \ .dropDuplicates(["user_id", "action", "ts"]) \ .filter(col("user_id").isNotNull()) \ .withColumn("amount", when(col("amount").isNull(), 0).otherwise(col("amount"))) \ .withColumn("dt", col("ts").substr(0, 10))

这段清洗逻辑做四件事：按业务键去重、过滤空用户、把缺失金额补零、从时间戳截取日期分区。注意dropDuplicates去重依赖字段选择，去重键必须能唯一标识一条记录，否则会把正常的多条行为误删。

5.2 多语言开发：Scala、Python、Java 三套 API 的协作边界

Spark-多语言开发.md 这份笔记解决了一个很现实的问题：团队里有人用 Python，有人用 Scala，最终任务怎么统一管理。Spark 3.0.1 本身是 JVM 应用，Scala 是原生 API，性能和功能覆盖最全；Python 通过 PySpark 调用，底层走 Py4J 桥接；Java API 在 3.0 里已经不那么常用，但维护老项目时还会遇到。

下面这份表格是我从笔记里归纳的选择标准：

用 PySpark 时最需要注意的是 Python UDF 的性能问题。每调用一次 Python UDF，数据都要在 JVM 和 Python 进程之间做序列化传输。能用pyspark.sql.functions内置函数解决的，就不要自定义 UDF，性能差距能达到数倍。Spark 3.0 的 Pandas UDF 可以缓解一部分性能问题，但要额外处理依赖包分发，入门阶段可以先了解，不急着深入。

5.3 从案例代码反推真实项目分层

综合案例看完了，接下来要把代码结构映射到真实项目的分层上。我见过太多人学完 Spark 之后，所有逻辑都堆在一个main方法里，能跑但是改不动。真实项目通常分四层：

• 接入层：负责读各种源数据，统一格式和 Schema
• 清洗层：空值处理、去重、格式规范化
• 计算层：指标逻辑、聚合规则
• 输出层：写 Hive 表、Parquet 文件或 Kafka 消息

// 分层写法的骨架示意 object UserActionJob { def main(args: Array[String]): Unit = { val spark = SparkSession.builder().appName("user_action_job").getOrCreate() val raw = DataLoader.loadUserActions(spark, args(0)) val cleaned = DataCleaner.clean(raw) val metrics = MetricsCalculator.computeDailyMetrics(cleaned) MetricsWriter.writeToHive(metrics, "dws.user_action_daily") spark.stop() } }

每个模块一个单例对象，职责单一，后面替换数据源或者改清洗规则时，只动对应模块就行。这个习惯比任何调优参数都重要：可维护的代码结构，才是大数据项目能持续迭代的基础。

6. 进阶验证：Spark 3.0 新特性和调优后的自查习惯

6.1 用 AQE 和动态分区修剪验证 3.0 新特性

Spark 3.0 相比 2.4 最明显的进步是自适应查询执行，简称 AQE。以前spark.sql.shuffle.partitions设成 200，所有 shuffle 都按 200 个分区跑，不管数据量多小。开了 AQE 之后，Spark 会在运行时根据实际 shuffle 数据量自动合并小分区，这个特性在 3.0 里默认开启，我一般会主动验证一下效果。

spark-submit --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.coalescePartitions.enabled=true --class com.example.MainJob /data/app/spark-demo.jar

验证方法很简单：跑完任务去 Web UI 的 Stages 页面看第二个 Stage 的 Task 数量，如果明显少于spark.sql.shuffle.partitions的设定值，就说明 AQE 生效了。另外尝试把spark.sql.adaptive.coalescePartitions.initialPartitionNum设成一个预期值，观察执行计划的变化，能加深对优化机制的理解。

6.2 任务上线前固定的五个检查动作

从那以后我每次提交 Spark 任务，都强制自己走一遍这套检查：先看输入路径是否存在、Schema 是否匹配；再看 SQL 执行计划里有没有全表扫描和过度 Shuffle；然后确认 checkpoint 目录没被其他任务占用；接着检查输出目录有没有历史残留，避免FileAlreadyExistsException；最后跑一个count()验证数据量符合预期。

这套习惯救过我不少次。最典型的一次是凌晨上线定时任务，忘了清理前一天留下的输出目录，结果任务一开始就失败，等到早上才发现。现在这些检查全写进启动脚本里，谁跑都一样。

把这套验证方法连同前面的调优参数、踩坑记录结合起来，再回去看 Spark-day01 到 Spark-day06 的笔记，会发现每一章的细节都串起来了。资源不难，难的是把坑提前踩完。

希望帮到你。

本文还有配套的精品资源，点击获取