简介一份Spark3.0入门到精通的全套学习资料内容基于2020年发布的Spark3.0.1稳定版覆盖Spark环境搭建、SparkCore、SparkStreaming、SparkSQL、StructuredStreaming、综合案例、多语言开发、3.0新特性与性能调优共9大章节面向零基础或刚接触大数据的学习者可系统掌握Spark核心开发技能。压缩包共244个文件主要包含217张png图表、10个md学习笔记、7个scala源代码及json配置整体大小86.9MBmd笔记对应1-8天课程配合scala代码和图表便于边学边练。目前已有609人学习下载内容按天递进、模块清晰从环境搭建到综合案例再到性能调优每个模块均配有对应笔记与代码综合案例覆盖日志分析、实时统计等典型场景多语言开发演示了不同API的调用方式性能调优则整理了常用参数与优化思路。既有原理讲解也有实战代码能帮助读者快速上手Spark3.0并理解流式处理与结构化查询的常见场景是一份适合自学和课堂教学的参考资料。1. 划重点这份 Spark 3.0 八天入门资料值钱的不是“精通”两个字我见过太多人把“大数据入门spark3.0入门到精通 1-8day 代码-笔记.zip”解压之后丢进网盘就再也没打开过。这资料本身不是问题问题在于大家把它当成了收藏夹而不是一条学习路线。八天的内容拆开看其实对应的是大数据开发最典型的三个阶段前三天搭环境、啃 RDD 和算子中间两天切到 DataFrame 和 Spark SQL最后几天用完整的数据分析案例把前面所有知识串起来。对准备走大数据开发路线但没摸过生产集群的初学者以及被 Excel 卡到想用 Spark 分析日志的数据分析师这套顺序都值得照着重走一遍。前提只有一个别只看笔记把每个代码块都敲进去。2. 先把环境立住版本匹配是第一步也是最容易翻车的一步2.1 安装组合怎么选JDK、Scala 和 Spark 3.0 的版本匹配是玄学也是底线Spark 3.0 的安装本身不复杂复杂的是“组合”。很多新手下载了最新版 JDK 17再配一个 Spark 3.0跑起来就报错然后开始怀疑人生。这不是代码问题是版本匹配问题。Spark 3.0 的预编译包是基于 Scala 2.12 编出来的你用 Scala 2.13 去写代码API 对得上但运行时不保证兼容。JDK 那边更直接Spark 3.0 官方支持 JDK 8 和 11你非要用 17某些反射调用会直接罢工。我一般会按这个组合装这也是这套 8 天笔记大概率默认的环境组件推荐版本备注JDK8u201 及以上Spark 3.0 对 JDK 8 支持最稳生产环境也大量用 8Scala2.12.x预编译包用 2.12不要自己配 2.13Spark3.0.x选择带 hadoop3.2 后缀的二进制包Hadoop3.2 或 2.7决定 tgz 包后缀纯本地学习选 hadoop3.2 即可下载之后解压路径我建议固定在/opt/spark后续所有配置都围绕这个目录展开。然后配置环境变量这一步决定了 spark-shell 和 spark-submit 能不能被直接调用export JAVA_HOME/usr/lib/jvm/java-8-openjdk-amd64 export SPARK_HOME/opt/spark export PATH$SPARK_HOME/bin:$SPARK_HOME/sbin:$PATH export PYSPARK_PYTHONpython3这四条环境变量是后面所有命令能跑起来的前提。JAVA_HOME必须指向 JDK 的安装根目录不是 bin 目录SPARK_HOME指向刚才解压出来的 spark 目录PATH里同时加了bin和sbinbin下面放的是 spark-shell、spark-submit 这类客户端命令sbin下面放的是 start-master.sh、start-slaves.sh 这类集群管理脚本。很多人只配了bin没配sbin结果脚本命令找不到还以为文件缺失。装完之后别急着往下走先敲一行spark-shell --version如果能看到 Spark 3.0.x 的版本信息说明 JAVA_HOME 和 SPARK_HOME 都生效了如果报 “Unsupported class file major version”立刻回去检查 JDK 版本。2.2 用 standalone 模式做最简 spark 集群搭建两条命令起一个可用的分布式环境入门阶段不建议直接上 YARN 或者 Kubernetes因为环境变量、资源队列、认证这些额外的东西会把注意力从 Spark 本身挪开。我建议先在 standalone 模式下手动搭一个最简 spark 集群搭建它包含了 Master 和 Worker 两个角色和真实分布式环境的结构一致但只需要一台机器就能演示。先在配置文件里指定当前机器的角色cd $SPARK_HOME/conf cp spark-env.sh.template spark-env.sh cp slaves.template slaves echo localhost slaves echo export SPARK_MASTER_HOSTlocalhost spark-env.sh注意 Spark 3.0 时代这个文件还叫slaves如果你用的是更新的版本文件名可能已经改成workers脚本也从start-slaves.sh变成了start-workers.sh。这一步的作用是告诉 Master 进程哪些机器是 Workerlocalhost表示本机参与计算。真正多机部署时这里写的是各台 Worker 的 IP 或主机名并且前提是各节点之间已经配好了 SSH 免密登录。接下来启动集群并验证进程$SPARK_HOME/sbin/start-master.sh $SPARK_HOME/sbin/start-slaves.sh jpsjps是 JDK 自带的小工具能列出当前机器上所有 Java 进程。正常情况下你至少能看到Master和Worker两个进程。然后打开浏览器访问http://localhost:8080Web UI 会显示 Master 的地址默认是spark://localhost:7077这个地址就是后面提交任务时的入口。如果 Web UI 有了但 Worker 显示为零大概率是slaves文件没生效或者 Worker 启动日志在$SPARK_HOME/logs下报了连接失败。看一眼日志就明白了不用瞎猜。2.3 第一个 Spark 程序用 spark-shell 跑 wordcount建立数据处理直觉环境起来之后不要急着写复杂业务先跑通最小闭环。spark-shell 是 Spark 自带的交互式命令行每敲一行代码立即在当前会话里执行非常适合入门阶段观察数据流。启动时显式指定 Master 地址$SPARK_HOME/bin/spark-shell --master spark://localhost:7077进入 Scala 交互界面后先准备一个本地文本文件再用最简单的 RDD API 做词频统计val rdd sc.textFile(file:///tmp/wordcount.txt) val counts rdd .flatMap(_.split( )) .map(word (word, 1)) .reduceByKey(_ _) counts.collect().foreach(println)这一段代码是 Spark 入门必写的“hello world”。sc.textFile读取文件返回的是 RDD[String]每一行是 RDD 里的一条记录flatMap(_.split( ))把每一行按空格拆成多个单词再拍平成一个单词列表map(word (word, 1))把每个单词变成键值对key 是单词本身value 是 1最后的reduceByKey(_ _)会在集群里按照 key 做分组求和相当于把相同单词的计数加起来。这里有个很重要的点collect()会把计算结果全部拉到 driver 端并以数组形式返回。数据量小的时候没问题生产环境数据量大的话千万不要 collect否则 driver 直接内存溢出。入门阶段用 collect 是为了快速看到结果但要养成“看到结果心里有数然后立刻去查 Web UI 的 Stage 执行图”的习惯。每个 Stage 的耗时、shuffle 数据量、执行器数量在那个 UI 里一目了然。3. 八天学习路径拆解从 RDD 算子到 Spark SQL 的代码节奏3.1 前三天集中火力吃透 RDD四个算子的数据流比语法重要翻过 Spark 3.0 入门资料的人都会发现前三天的笔记里几乎全是 RDD 的例子。这不是为了怀旧而是因为算子的执行模型——哪些操作在 executor 上跑、哪些操作会产生 shuffle——才是理解 Spark 的底层骨架。RDD 表面上是一堆 API但实际上只有变换和动作两类操作map、filter、flatMap、reduceByKey都是变换它们只是定义计算流程不触发真正执行而collect、count、saveAsTextFile这类动作才会启动任务。我建议前三天只用 spark-shell 做实验反复敲这四个算子的组合val rdd sc.parallelize(Seq(hadoop spark, spark flink, hadoop hive)) val words rdd.flatMap(_.split( )) val wordCounts words .map(word (word, 1)) .reduceByKey(_ _) wordCounts.sortBy(_._2, ascending false).collect().foreach(println)这段代码把3.1里那个词频统计升级了加入了sortBy(_._2, ascending false)按 value 降序排列这样能直接看到出现频率最高的单词。parallelize是把内存里的本地集合变成 RDD 的最低成本方式适合写 demo但真实项目里几乎不用因为数据永远不会在内存里等着你。需要注意reduceByKey和groupByKey的区别。reduceByKey会在 map 端先做一次预聚合然后把聚合后的结果通过网络传给 reduce 端groupByKey则是把所有原始键值对原封不动地传到下游。数据量一大groupByKey的 shuffle 数据量可能是reduceByKey的几倍甚至几十倍数据倾斜的风险也更高。8 天笔记里如果反复出现这两个算子值得多花半天时间用一个大文件对比一下两者的执行时间和 Shuffle Size。前三天最容易犯的错误是把 notebook 当 IDE。直接把 scala 文件丢进 spark-shell 里跑没问题但你会失去一个关键能力通过 UI 观察每个 Stage 的输入输出大小。入门阶段建议敲一段、看一段 UI对“代码”和“任务”之间的对应关系建立起条件反射。3.2 第四和第五天从 RDD 切到 DataFrame先记下“更快”的真正原因到第四天笔记里会出现 DataFrame 和 SparkSession。一开始我也以为 DataFrame 只是“带列名的 RDD”直到看完执行计划才明白差距在哪DataFrame 有 schemaSpark 能在逻辑计划阶段做列裁剪和谓词下推把不需要读的列和行过滤掉剩下的由 Catalyst 优化器生成高效的物理执行计划。这比 RDD 的逐算子执行要聪明得多。看这一段典型代码val spark SparkSession.builder() .appName(dataframe-demo) .master(local[2]) .getOrCreate() import spark.implicits._ val df Seq( (订单10001, 华东, 299.0), (订单10002, 华北, 59.9), (订单10003, 华东, 159.0) ).toDF(order_id, region, amount) df.show() df.filter($region 华东) .groupBy($region) .agg(sum(amount).as(total)) .show()SparkSession.builder()是 Spark 2.0 之后的统一入口3.0 里依然如此。.master(local[2])表示在本地用两个线程模拟 executor上手阶段不想折腾集群的时候可以直接 local 模式退到第 2 章搭好的 standalone 集群时把这行改成.master(spark://localhost:7077)就行。import spark.implicits._提供了字符串到列对象的隐式转换所以后面能写$region这种语法它和col(region)等价。agg(sum(amount))在 DataFrame API 里是聚合动作功能相当于 SQL 里的SUM。对比 RDD 版本的代码DataFrame 版本最直观的差异是代码量少了且每一步都能通过.printSchema()查看数据类型。如果笔记里有一步“把 CSV 读成 DataFrame 再做 where 过滤”的练习建议跑完之后执行一条df.explain(true)看物理计划里是不是多了PushedFilters和PruneFilterProject。看到这两个词你就理解了 DataFrame 为性能做的底层优化以后跟别人聊 Spark 也不只是“语法我会”了。3.3 第六到第八天把知识汇成一个完整的数据分析案例JSON 读取是标准动作到了最后三天资料通常会给出一个综合案例比如订单数据分析。这个阶段的核心不是学新算子而是把前面拆开的零件组装成一条流水线读数据、清洗、聚合、输出。我见过最典型也最容易翻车的场景是把 JSON 文件直接当成普通文本去读然后发现 JSON 文件一行可能是嵌套的 JSON 对象split 出来的东西没法直接用。正确做法是用spark.read.jsonSpark 会自动推断 schemaval orders spark.read.json(file:///tmp/orders.json) orders.printSchema() orders.createOrReplaceTempView(orders) val result spark.sql( SELECT region, COUNT(1) AS order_cnt, ROUND(SUM(amount), 2) AS total_amount FROM orders GROUP BY region ORDER BY total_amount DESC ) result.show() result.coalesce(1) .write.mode(overwrite) .parquet(file:///tmp/order_report)spark.read.json会自动推断每个字段的类型但注意如果一张表里同一个字段有时候是字符串有时候是数字推断结果会偏向宽泛类型后面算 SUM 的时候可能报错。所以实际业务里读进来先printSchema()检查推断结果再决定是否需要显式 cast。createOrReplaceTempView是把 DataFrame 注册成临时表之后就能用我们熟悉的 SQL 语法操作了。对大多数从 SQL 背景转过来的同学spark.sql绝对是舒适区。coalesce(1)用来把结果合并成一个分区再写出去这样最终输出的 parquet 文件只有少量文件而不是一批碎文件。入门阶段为了演示方便可以这样做生产环境不要这么干因为单分区写入会丢掉并行度大结果集反而会变慢。这里还顺手用到了write.mode(overwrite)这个参数直接决定了写文件时如果目录已存在是覆盖还是报错默认是报错测试环境经常要用overwrite生产环境则要根据业务慎重选择。4. 绕不开的三个概念分区、shuffle、spark 内存决定了任务快不快4.1 分区数到底怎么定它同时决定了并行度和输出文件数资料跑完之后很多学习者的成就感停留在“代码能跑通”但一旦被问到“你这个作业起了几个 task”就答不上来。分区数是 Spark 调优的第一道门槛它决定了一次作业被切成多少块并行执行。spark.default.parallelism在本地模式下是 CPU 核数在集群模式下由 executor 数量和 cores 决定spark.sql.shuffle.partitions则专门管 shuffle 后生成的 RDD 分区数默认 200。分区数不是越大越好。分区过多每个分区的数据量太小task 启动和调度的开销可能超过计算本身的耗时分区过少集群资源闲置大任务跑成串行。我见过一个坑某个案例里用spark.sql.shuffle.partitions5000跑一个很小的 groupBy结果 5000 个空 task 排队作业时间反而从 2 分钟涨到 15 分钟。入门阶段的判断基准很简单让每个分区的数据量落在 100MB 左右再在这个范围上下调整。spark-submit \ --class com.example.OrderReport \ --master spark://localhost:7077 \ --conf spark.sql.shuffle.partitions200 \ --conf spark.default.parallelism8 \ order-report.jar--conf的值对所有作业生效优先级高于代码里的 SparkConf。这两行的含义是shuffle 阶段切成 200 个分区默认并行度设为 8相当于让 8 个 core 同时干活。如果你发现作业大部分时间浪费在“task 调度”上而每个 task 实际计算时间不足一秒把分区数调小如果 GC 频繁但 CPU 没跑满通常不是分区问题往下看内存。4.2 spark 内存模型executor 内存和堆外内存的配置讲究Spark 的内存管理是入门阶段最容易被当成玄学的内容。实际上executor 的内存被分成了三块执行内存Execution、存储内存Storage和其他。Spark 3.0 里执行内存和存储内存是统一管理的可以互相借调但这意味着如果你把spark.executor.memory设得过大而 driver 或集群物理内存不够作业可能在执行阶段直接报 OOM反过来设得过小shuffle 和缓存又不够用。我一般会给一份保守的起步配置跑通之后再根据 Web UI 的 Executors 页面去调整参数名默认值建议spark.executor.memory1g单机学习设 2g-4g生产按每 executor 4g-8g 起步spark.executor.cores1通常和内存一起定比如 4g 配 2-4 核spark.driver.memory1gdriver 只做任务调度和结果的汇总不用太大spark.memory.fraction0.6执行存储占比默认 0.6GC 频繁时可调回 0.5spark.sql.shuffle.partitions200见上一节大 shuffle 场景要调大配置这些参数有两条路写死在代码的 builder 里或者提交任务时用--conf传入。在这套 8 天笔记里我建议直接改成可复用的 Python 提交脚本每台机器改一两个变量就能复用spark SparkSession.builder \ .appName(spark-memory-tuning) \ .master(spark://localhost:7077) \ .config(spark.executor.memory, 4g) \ .config(spark.executor.cores, 2) \ .config(spark.driver.memory, 2g) \ .config(spark.sql.shuffle.partitions, 200) \ .config(spark.serializer, org.apache.spark.serializer.KryoSerializer) \ .getOrCreate()最后那行.config(spark.serializer, ...)是很多教程里没细说的点默认的 Java 序列化器方便但体积大换成 Kryo 序列化能显著降低网络传输和内存占用。但 Kryo 需要注册类如果没注册某些自定义对象会报ClassNotFoundException或者 “Kryo serialization failed” 之类的问题。入门阶段跑官方自带的类型直接用 Kryo 问题不大但一旦开始用自己的样例类就要小心这个坑。4.3 shuffle 既是磁盘操作又是网络操作别再忽略它写 Spark 作业的人最常低估的就是 shuffle。所谓 shuffle是把分布在各个 executor 上的数据按照 key 重新分发到对应的下游任务节点这个过程一定会落盘也一定会有网络传输。数据量大时shuffle 写盘量甚至比最终计算结果大一个数量级。判断一个任务有没有 shuffle最直接的方法是看 Web UI 那条 DAG 图里有没有标记为 “Shuffle Read” 和 “Shuffle Write” 的阶段。产生 shuffle 的典型算子包括reduceByKey、groupByKey、join、distinct。能避免就避免比如先做一轮过滤把不参与关联的数据提前清掉再去 join。我在实际处理里经常用的一条经验是不管代码多复杂先从df.explain()的输出里数一下出现Exchange的次数每出现一次Exchange就是一次 shuffle。如果一次查询里出现三次以上的Exchange基本可以先停下来想想能不能通过调整 SQL 的关联顺序、或者在数据源层面做预处理来减少次数。这也是后面验证调优效果的最基础手段。5. 避坑照着笔记敲代码翻车最多的五个瞬间5.1 跑 wordcount 报 No Space Left on Device但磁盘明明还有 100GB现象spark-shell里跑一个大文件词频统计执行几个 stage 之后直接报java.io.IOException: No space left on device检查/df -h发现磁盘并没有满。原因Spark 在执行 shuffle 时会在spark.local.dir指定的临时目录里写中间数据。默认的临时目录是/tmp而很多生产机器挂载的/tmp目录配额很小和根目录/不在一个文件系统里。解决查看$SPARK_HOME/conf/spark-defaults.conf把spark.local.dir指向一个有足够空间的大目录一般配置成/data/spark-tmp或$SPARK_HOME/tmp然后重启集群。配置后可以用df -h确认挂载点容量也可以用du -sh查看 worker 节点临时目录的实际占用确认 shuffle 数据已经写到新路径。5.2 用 file:/// 读本地文件报文件不存在但文件确实在那里现象在 spark-shell 里sc.textFile(file:///tmp/sample.txt)报No such file or directory可是 shell 里ls /tmp/sample.txt明明看得到。原因spark-shell 里 shell 和 executor 进程可能跑在不同节点或者分布式模式下 executor 的本地文件系统和 driver 本地文件系统不是同一套。只要任务被分发到远程 worker 上的 executorexecutor 进程就去它自己那台机器的本地目录找文件自然找不到 driver 机器上的文件。解决入门阶段最省事的办法是改用 resource 方式提交把数据文件打成 jar 包的一部分用spark-submit --files sample.txt随任务一起分发。或者更简单先把文件上传到 HDFShdfs dfs -put /tmp/sample.txt /data/代码里路径写成hdfs://localhost:9000/data/sample.txt。搞清楚 driver 端和 executor 端的区别就是搞清楚“把文件放在哪里才靠谱”这个问题的本质。5.3 Dataset.show() 只显示 20 行我以为数据丢失了现象跑完聚合以后执行result.show()输出只有 20 行还打印一段only showing top 20 rows就怀疑是哪一步数据被过滤掉了。原因这是 Spark 的自我保护机制。show() 方法默认显示最多 20 行避免在交互终端一次性打印海量数据造成日志刷屏或终端卡死。数据本身没有被截断它只是被故意“挡住”了。解决在 show 里传参数改变行数比如result.show(100)显示 100 行或者result.show(200, false)表示显示 200 行且不截断超长字段。如果要用完整数据应该走collect或直接输出到文件而不是和 show 的死活较劲。5.4 作业 OOM 了但到底是谁的内存溢出日志没说现象作业运行时日志里频繁出现java.lang.OutOfMemoryError但没有清晰提示是 driver 还是 executor 的问题。原因Spark 的 OOM 可能有三种来源driver 端拿结果拿爆了executor 的堆内存被大数据量撑爆了以及堆外内存耗尽。collect最常在 driver 端炸groupByKey容易导致某个 executor 的缓冲过大而频繁cache数据则可能吃光堆外内存。解决第一件事是确认错误出现的阶段。如果发生在最后的 action 阶段基本是 driver 内存不够调大spark.driver.memory如果发生在 shuffle 读取过程中优先检查分区数和spark.executor.memory。正经的排查路径是去 Web UI 的 Executors 页面看每个 executor 的 Memory 列是接近内存上限的红色还是正常绿色再决定调参方向不要一上来就把内存翻倍。5.5 reduceByKey 和 groupByKey 用错数据倾斜导致任务卡死现象作业里某个 stage 卡在 99% 不动某个 task 运行了几十分钟还没结束其他 task 早就完成。原因reduceByKey 和 groupByKey 都会产生 shuffle但 shuffle 后数据分区的分布取决于 key 的散列情况。如果某几个 key 特别集中那一个分区里的数据量会远超其他分区单个 task 就会成为整个作业的短板。解决先看 Web UI 里每个 stage 的 task 时间分布图如果有一条长尾基本就是倾斜了。缓解的手段很多入门阶段先试两个一是把groupByKey换成reduceByKey让 map 端先合并一部分数据二是给倾斜的 key 加随机前缀把它们打散比如map(k ((k, Random.nextInt(10)), v))聚合完再去掉前缀做第二次聚合。后者属于经典的两阶段聚合也是后续调优实战里一定会遇到的套路。6. 别急着“精通”先学会三种验证方法再谈进阶资料里的“精通”是营销词真正能检验你水平的不是跑通了几段代码而是面对一个新数据文件时你能不能快速判断“它会被怎么计算、瓶颈在哪里”。我建议用三种验证方法替代“感觉代码能跑”。第一种看 Web UI 的 Stage 耗时分布。跑任何一个作业之后打开 Master 的 8080 端口进入 Application 页面点进具体 stage看 Input、Shuffle Read、Shuffle Write 三个指标的数值。如果 Shuffle Read 远远大于 Input说明很多数据没必要跑到下游去代码里大概率存在过早展开的聚合操作。第二种学会读执行计划df.filter($amount 100) .groupBy($region) .count() .explain(true)explain(true)会打印完整执行计划从逻辑计划到物理计划都能看到。物理计划里如果出现了PushedFilters说明过滤条件下推到数据源了如果只有Scan和HashAggregate说明查询结构非常紧凑。用这个命令替代“看着代码猜性能”的黑匣子思考方式是新手的第一个分水岭。第三种同一份数据冷热各跑一遍对比耗时。Spark 有cache机制数据一旦缓存后续查询不会重新从磁盘读取。把同一段聚合脚本执行两次第一次跑完执行df.cache()第二次再看耗时如果第二次显著变快说明缓存生效了如果两次差不多可能是数据量太小或者缓存被spark.memory.fraction里存储部分的低位挤掉了。用自己的测试换个思维理解缓存比背十条调优口诀管用得多。我自己最后悔的就是当年想把 8 天的内容压缩成 3 天看完结果环境装了一周还没跑通。后来带人入门我定了一条死规矩Day 1 只做环境Day 2 才允许在 spark-shell 里敲第一行 wordcount。慢就是快这个道理在分布式系统里面格外真实。希望帮到你。本文还有配套的精品资源点击获取