★ FEATURED ARTICLE “Spark---小练习”这个标题乍看平平无奇但我太熟悉这个状态了很多刚学Spark的人跑完官方文档里的wordcount就以为自己“会了”结果一碰到真实数据就露馅——RDD该用map还是flatMapgroupByKey为什么越跑越卡本地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用户可以直接用PySparkJava版本的影响会小一些但同样要注意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(secondsrandom.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/UVPV就是每小时多少条浏览记录UV就是每小时多少个不同用户访问。热门商品Top10按浏览、加购、下单加权打分看哪些商品综合热度最高。下单用户的平均浏览商品数先筛出至少下过一次单的用户然后统计这些用户浏览过多少种不同商品再求平均值。选这三个指标不是随手写的它们分别对应Spark里最核心的能力时间维度拆分和去重计数、分组聚合加排序取TopN、多条件关联和子查询。如果你把这三种场景都跑通了日常工作中八九成的数据分析需求基本都能覆盖到。而且这三个问题的复杂度是递增的第三题尤其能逼你思考“怎么跨数据集做匹配”这是从“会算子”到“会组合算子”的关键一步。3. 先用RDD跑通第一版算子选择里的门道3.1 map和flatMap最容易绕晕的两个算子读文件时textFile返回的是一行一行的RDD[String]接下来要按制表符把每个字段拆开。这里就碰到第一个选择用map还是flatMap。最简单的区分方式是map对每个元素只输出一个元素flatMap对每个元素可以输出零个或多个元素。比如按制表符分割一行日志经过处理后仍然是一条记录这时候用mapval 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最简单按小时分组累加1val 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数据量大概280MBGC时间占总运行时间的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 常见报错和处理方法练习过程中遇到报错是常态我这里整理了几个出现概率最高的每个我都亲自踩过报错现象根本原因解决办法Failed to locate the winutils binaryWindows下缺少Hadoop临时工具配置HADOOP_HOME并放好winutils.exeFileAlreadyExistsException输出目录已存在跑之前手动删掉输出目录Java heap spaceDriver或Executor内存不够调大driver-memory或检查是否存在数据倾斜Task not serializable闭包里捕获了不可序列化的对象把变量提取成静态常量或局部值UnsupportedOperationException: empty collection对空数据集做了聚合操作先做空值判断或者用firstOption兜底这些报错的共同点是表面症状五花八门实际原因往往只有两三个。所以我每次排查都会先问三个问题数据是不是空的分区数是不是不合理闭包里是不是引了不该引的东西问完这三个八成问题都能定位。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改成了什么样。这几步做到位了你才算真的把这个练习“吃透”了。
阅读完成 · 觉得有帮助？
