Spark小练习：从环境配置到SQL优化，吃透一次日志分析

“Spark---小练习”这个标题，乍看平平无奇，但我太熟悉这个状态了：很多刚学Spark的人，跑完官方文档里的wordcount，就以为自己“会了”，结果一碰到真实数据就露馅——RDD该用map还是flatMap？groupByKey为什么越跑越卡？本地IDE里好好的，换到集群就各种报错。这篇就拿一个小练习当解剖样本：某电商平台一天的用户行为日志，大概几百万条，去算每小时PV/UV、热门商品Top10、下单用户的平均浏览商品数。这个练习规模不大，但足够把环境配置、算子选择、SQL优化、作业调优完整走一遍。适合刚学完Spark基础、想动手练点真东西的人，也适合那些“跑完wordcount就不知道下一步该做什么”的朋友。

1. 先搭好本地环境：版本选型和最容易翻车的三个细节

1.1 版本别追新，匹配才是硬道理

很多新手喜欢装最新版，觉得新版本功能多、性能好，但在Spark这里，版本匹配的价值远大于版本新。我一开始在本地搭练习环境的时候，照着当时最新的教程配了高版本的Spark，结果和Scala编译版本对不上，程序一跑就报方法找不到，排查了半天才发现是某个库的二进制兼容问题。

我自己练手用的是这么一套组合：Java 8、Spark 3.3.2、Scala 2.12、sbt 1.9。这套组合我用了很久，网上资料多，踩坑的人也多，遇到问题基本都能搜到答案。如果你用Java 17跑Spark，经常会碰到反射访问模块被拦截的报错，还得额外加一堆JVM参数，学习阶段搞这些东西纯属分散注意力。同样，Scala版本也要和Spark发行版匹配，Spark 3.3对应的默认Scala版本是2.12，你如果手滑用了2.13编译，运行时就会莫名奇妙报NoSuchMethodError——这就是版本不对齐的典型症状。

Python用户可以直接用PySpark，Java版本的影响会小一些，但同样要注意Spark版本和Python版本的兼容关系，别贪新。练习阶段选一套“老而稳”的组合，把精力留给后面真正核心的算子逻辑，才是正确姿势。

1.2 本地模式跑通，这三个配置最容易漏

第一个是setMaster("local[4]")。这个词要解释一下：local[n]表示用本地机器的n个线程模拟分布式执行。为什么要指定4？因为一个核跑一个任务，本地机器如果8核16G，我一般设置4到8个线程。设成local[*]也行，表示自动按机器核数来，但要注意，如果你电脑本身内存就紧张，核数全开反而让每个任务分到的资源更少。在代码里这样写：

val spark = SparkSession.builder() .appName("spark-basic-practice") .master("local[4]") .config("spark.driver.memory", "2g") .getOrCreate()

第二是Windows用户最容易踩的坑：Failed to locate the winutils binary in the Hadoop binary directory。这个报错跟Spark本身没关系，是本地程序想调Hadoop的临时文件工具，但Windows没有这个二进制。解决办法很直接：下载对应Hadoop版本的winutils.exe，放到一个固定目录，然后把HADOOP_HOME环境变量指过去。如果嫌麻烦，直接用Linux虚拟机或者云主机跑练习，能省掉这一堆环境折腾，但对Windows用户来说，解决这个报错本身也算一种练习。

第三是输出目录的问题。Spark写结果时不会自动覆盖已有文件，如果你跑第二次，输出目录还在，它会直接报FileAlreadyExistsException。本地练习我通常会在代码里先判断一下，或者干脆在跑之前手动删掉输出目录：

rm -rf output/sql_top10

这三个配置，我几乎每次带人入门都会反复强调。环境搭建花掉的时间，往往比写练习代码本身还要多，提前知道这些坑，能让你把时间花在刀刃上。

2. 小练习怎么设计，才不算是“白练”

2.1 先造一批干净的数据集

有人喜欢拿公司的真实生产数据练手，我的建议是不要。生产数据可能有敏感字段、字段含义不明、量级太大，跑一次要等半天，练习时目标完全被干扰。模拟数据虽然“假”，但它干净、可控、字段含义清晰，能让你一门心思关注Spark本身的逻辑。

我练习用的是这样一个场景：某平台的用户行为日志，每行一条记录，字段分别是时间、用户ID、商品ID、行为类型。行为类型就三种：view、cart、order，分别表示浏览、加购、下单。字段间用制表符分隔。生成脚本用Python写很简单：

import random import datetime users = ["u_%06d" % i for i in range(50000)] products = ["p_%04d" % i for i in range(280)] # 浏览概率远大于加购和下单，更贴近真实场景 actions = ["view"] * 7 + ["cart"] * 2 + ["order"] * 1 base = datetime.datetime(2024, 5, 12, 0, 0, 0) with open("logs.tsv", "w") as f: for _ in range(5000000): ts = base + datetime.timedelta(seconds=random.randint(0, 86399)) uid = random.choice(users) pid = random.choice(products) act = random.choice(actions) f.write(f"{ts:%Y-%m-%d %H:%M:%S}\t{uid}\t{pid}\t{act}\n")

这样跑下来大概生成500万条记录，文件400多MB。在8核16G的笔记本上，这个量级既能体现计算引擎的调度开销，又不会等太久。如果你电脑配置低，可以改成100万条，后面的性能对比同样明显。

2.2 三个指标为什么这样设计

这次小练习我定了三个目标：

• 每小时PV/UV：PV就是每小时多少条浏览记录，UV就是每小时多少个不同用户访问。
• 热门商品Top10：按浏览、加购、下单加权打分，看哪些商品综合热度最高。
• 下单用户的平均浏览商品数：先筛出至少下过一次单的用户，然后统计这些用户浏览过多少种不同商品，再求平均值。

选这三个指标不是随手写的，它们分别对应Spark里最核心的能力：时间维度拆分和去重计数、分组聚合加排序取TopN、多条件关联和子查询。如果你把这三种场景都跑通了，日常工作中八九成的数据分析需求基本都能覆盖到。而且这三个问题的复杂度是递增的，第三题尤其能逼你思考“怎么跨数据集做匹配”，这是从“会算子”到“会组合算子”的关键一步。

3. 先用RDD跑通第一版：算子选择里的门道

3.1 map和flatMap，最容易绕晕的两个算子

读文件时，textFile返回的是一行一行的RDD[String]，接下来要按制表符把每个字段拆开。这里就碰到第一个选择：用map还是flatMap。

最简单的区分方式是：map对每个元素只输出一个元素，flatMap对每个元素可以输出零个或多个元素。比如按制表符分割，一行日志经过处理后仍然是一条记录，这时候用map：

val rdd = sc.textFile("logs.tsv").map { line => val arr = line.split("\t") (arr(1), arr(2), arr(3), arr(0).substring(0, 13)) }

如果你把分割后的结果直接当成一个数组传出去，比如line.split("\t")返回的是Array，因为数组里的每个元素都会被单独当成一条记录，这时候就得用flatMap，它会把数组“摊平”。我看过很多新手在这里写错：map之后发现后续算子拿到的不是字段，而是一个数组，然后又开始做各种奇怪的下标处理，绕了半天。其实第一步就错了。flatMap还有一个实用场景是数据过滤，比如想丢掉空行：

rdd.flatMap(line => if (line.isEmpty) None else Some(line))

所以我的经验是：先想清楚“这个算子对每个记录应该产生几条输出”。一对一用map，一对多或者零对多就用flatMap。

3.2 reduceByKey和groupByKey，选错就是几倍的性能差距

接下来算每小时PV和UV。PV最简单，按小时分组，累加1：

val hourlyPv = rdd.map(r => (r._4, 1L)).reduceByKey(_ + _)

UV稍微绕一点，得先把用户去重，再按小时计数：

val hourlyUv = rdd.map(r => (r._4, r._1)) .distinct() .map(r => (r._1, 1L)) .reduceByKey(_ + _)

很多教程在这里会强调：能用reduceByKey的地方，不要用groupByKey。原因是reduceByKey会在map端先做一次局部合并，比如同一个小时内的PV，先在本地节点上把多个1加成一个数，再通过网络shuffle汇总；groupByKey则不做任何预合并，直接把同一key的所有value全量拉到一个节点上，再在reduce端处理。网络传输的数据量差了好几倍。练习数据量小看不出来，一旦数据上亿，这个差距就是几分钟和几十分钟的差别。

第三题在RDD里写起来最痛苦：要先找出所有下单用户，再过滤出这些用户的浏览记录，然后按用户去重商品，最后求平均。我用了一个很直接但“坏味道”很重的写法：

val orderUsers = rdd.filter(_._3 == "order") .map(r => (r._1, 1L)) .reduceByKey(_ + _) .map(_._1) .collect() .toSet val browseCount = rdd.filter(r => r._3 == "view" && orderUsers.contains(r._1)) .map(r => (r._1, r._2)) .distinct() .map(r => (r._1, 1L)) .reduceByKey(_ + _) .map(_._2) .mean()

这里把下单用户的ID集合通过collect()收回到Driver端，再广播到每个任务里去过滤。在几百MB的小数据量练习里它能跑，但放在生产环境，几千万用户的下单集合直接塞进Driver内存就是一场灾难。这个问题的根因是：你下意识用了单机编程的思维，在一个分布式引擎里强行把数据搬回“主节点”。这个坑，几乎每个从单机Python转过来的新手都要踩一次。

3.3 第一版跑出来的结果，为什么这么慢

第一版在8核16G的笔记本上，跑这500万条数据，花了4分多钟。点开UI一看，Shuffle数据量大概280MB，GC时间占总运行时间的12%左右，也就是说每跑十分钟有超过一分钟在垃圾回收。

为什么会这样？关键就在distinct()和那个collect().toSet。前者引入了一次全量shuffle，后者虽然只收集了一次，但让后续的所有map都带上了Driver端变量查找的开销。另外RDD代码里大量中间RDD对象分配，也让JVM的GC压力变大。第一版的意义是帮你把流程跑通，验证结果正确，但看到这些数字你就该明白：后面有优化的空间。

4. 用DataFrame和SQL重构：为什么代码短一半还跑得快

4.1 DataFrame背后到底做了什么优化

不少人觉得DataFrame只是“换了个写法”，性能未必更好，这种想法大错特错。DataFrame走上的是Catalyst优化器加Tungsten执行引擎这条路。Catalyst会把你的操作解析成逻辑执行计划，然后做一堆规则优化，比如谓词下推、列剪裁；Tungsten则用内存二进制格式管理数据，减少JVM对象开销。同样的逻辑，RDD里你真刀真枪写的map和reduce，其实和写SQL相比并没有“更底层”，反而丢失了优化机会。

所以在跑过RDD版本之后，我建议立刻用DataFrame重写一遍。这不只是语法转换，而是让你直观感受到“同样的数据，同样的机器，为什么换一个API就能快这么多”。这种体感比任何理论讲解都有效。

4.2 迁移步骤和几个核心坑

读取数据、注册临时视图，这一步很简单：

val df = spark.read .option("sep", "\t") .option("header", "false") .csv("logs.tsv") .toDF("ts", "user_id", "product_id", "action") df.createOrReplaceTempView("logs")

接下来三个指标直接写SQL：

-- 每小时PV/UV SELECT substr(ts, 1, 13) AS hour, COUNT(*) AS pv, COUNT(DISTINCT user_id) AS uv FROM logs GROUP BY hour ORDER BY hour; -- 热门商品Top10 SELECT product_id, SUM(CASE action WHEN 'view' THEN 1 WHEN 'cart' THEN 5 WHEN 'order' THEN 10 ELSE 0 END) AS score FROM logs GROUP BY product_id ORDER BY score DESC LIMIT 10; -- 下单用户的平均浏览商品数 SELECT AVG(cnt) FROM ( SELECT user_id, COUNT(DISTINCT product_id) AS cnt FROM logs WHERE action = 'view' AND user_id IN (SELECT DISTINCT user_id FROM logs WHERE action = 'order') GROUP BY user_id ) t;

这几段SQL，一看就懂。最需要注意的坑是：默认spark.sql.shuffle.partitions是200。这个值是为大集群设计的，本地8核16G的机器上，跑500万条数据根本不需要200个reduce分区。每个task处理的数据太少，调度开销反而比数据处理本身还大。我把它调成8之后，任务运行时间从一分钟左右直接降到几十秒。

spark.conf.set("spark.sql.shuffle.partitions", 8)

还有一个写结果的小技巧，用coalesce(1)控制结果文件数量。不控制的话，每个分区任务都会落一个文件，最后可能生成十几个几十KB的小文件，又碎又难处理。练习阶段用coalesce(1).write.csv("output/xxx")最方便。

4.3 那些“能跑但很危险”的坏味道

SQL写起来短，但不代表可以乱写。我见过很多练习作品直接裸用COUNT(DISTINCT user_id)，在小数据量上没毛病，但在大量数据且user_id分布不均匀时，这个去重操作会十分吃力，极端情况下直接内存溢出。生产中更稳妥的做法是两阶段聚合：先按某个中间维度做局部去重计数，再对结果求和；或者用近似去重函数approx_count_distinct。练习数据虽然不需要这么做，但你得知道这个边界在哪。

另一个坏味道是把时间当字符串随便截取。substr(ts, 1, 13)在数据规整时能得到小时，但一旦数据跨了时区或者时间字段格式变了一点，结果就奇奇怪怪。生产上建议统一转成时间戳，再格式化成小时。本地练习怎么简单怎么来，但心里要留根弦：这不是通用做法。

跑完DataFrame版本之后再看耗时：RDD版4分多钟，DataFrame版几十秒，代码行数也从60多行降到20多行。这就是我为什么总对新手说，别把RDD当成首选，它更适合做底层算子研究，而不是日常数据分析。

5. 运行、排查与作业调优：文档里不会写的三件事

5.1 从Spark UI读懂一次作业的“病情”

本地作业在跑的时候，访问http://localhost:4040就能看到Spark UI。我第一次跑RDD版本时，对着一堆时间线发呆，完全不知道看什么。后来我总结了一个最简单实用的观察路径：先看Executors页签里的GC时间占比，再看SQL/Stage页签里的Shuffle数据量。

如果GC时间占总时间10%以上，说明你在大量地创建对象，典型的元凶就是RDD反复做distinct和map时产生的中间对象。如果某个Stage的Shuffle Write特别大，说明前一步的聚合或过滤下推没做好。这些都是可以量化的信号，而不是靠感觉排查。当你把spark.sql.shuffle.partitions从200改成8时，你会在UI里看到Task数量直线下降，这个直观反馈比任何教程都更有说服力。

5.2 常见报错和处理方法

练习过程中遇到报错是常态，我这里整理了几个出现概率最高的，每个我都亲自踩过：

这些报错的共同点是：表面症状五花八门，实际原因往往只有两三个。所以我每次排查，都会先问三个问题：数据是不是空的？分区数是不是不合理？闭包里是不是引了不该引的东西？问完这三个，八成问题都能定位。

5.3 一个练习的三种收尾方式

练习写完之后，建议不要只在IDE里点运行，试着用spark-submit提交一次。这一下就能把“开发环境写代码”和“提交作业跑任务”这两件事衔接起来。

spark-submit \ --master local[4] \ --driver-memory 2g \ --class com.example.PracticeJob \ spark-basic-practice.jar \ logs.tsv output/

程序里用args(0)接收输入路径，args(1)接收输出路径，这样这套练习就变成了一个可以重复运行的小工具，而不是写死在代码里的演示。更进一步，你可以把第三题的SQL改成用窗口函数实现，或者引入一个模拟数据流，把批处理练习延伸到微批处理的方向。这个练习本身不大，但把它当一个“可插拔的实验台”，后续很多Spark新特性都可以拿到这里来试。

最后分享一个我个人的体会：Spark最大的门槛不是语法，而是“跑完之后你根本不了解你的作业到底干了什么”。下次再练任何一个Spark小练习，跑完之后别急着关页面，去UI把每个Stage的Shuffle数据量看一眼，把RDD版本和DataFrame版本的时间线对比一下，再用explain()打印一下物理执行计划，看看SQL被Catalyst改成了什么样。这几步做到位了，你才算真的把这个练习“吃透”了。