如果你最近在学 Spark又想找一个不绕弯、不烧集群的实战练习这一组 Spark 小练习应该正合你意。我最近刚把它完整过了一遍用 PySpark 读一份本地生成的销售订单 CSV接着做数据清洗、去重和日期规整再用 DataFrame API 与 Spark SQL 分别算出“每个月、每个品类的订单量、销售额、平均客单价”最后把结果写出并回读校验。整个流程不依赖分布式集群单机就能跑也不需要你提前会 Scala。它没有多高深但对靠刷题入门的人来说能一次性补上最关键的“从文件到 DataFrame再从 DataFrame 到结果报表”的完整手感。这个小练习解决的是一个很典型的痛点很多人把 RDD 算子背得滚瓜烂熟可真给一份 CSV却不知道怎么下手也有人只会写 Spark SQL但完全不理解 DataFrame 算子与 SQL 的对应关系。做完这组练习你会对groupBy、agg、withColumn、filter这些概念有具体体感也能看懂 Spark UI 里最简单的 Job 和 Stage 长什么样。如果你是刚学完 Spark 基础的学生或者准备大数据开发面试但缺实操的初学者可以照着下面的方案完整走一遍如果已经能用 Spark 做生产任务也可以重点看后面的问题排查部分。1. 动手前先拆解这个小练习到底练什么1.1 练习目标不是搭集群而是跑通一条完整的 Spark SQL 数据链路我第一次做这个练习时差点被“分布式”“集群”这些词带偏想着要不要先搭一个三节点环境。后来我意识到对一个以学习为目的的小练习来说最关键的不是机器数量而是你能不能把一条数据处理链路完整地走通。所谓链路在这里就是五个环节读入外部数据、观察数据形态、做清洗转换、执行聚合统计、把结果写回文件系统。我把每个环节对应到具体动作第一步用spark.read.csv把本地 CSV 变成DataFrame第二步用printSchema和show看字段类型与样例数据第三步做状态过滤、订单去重和日期字段拆分第四步用groupBy与agg算出每个品类每个月的订单量和销售额第五步把结果写成一个 CSV 文件再重新读回来验证。五个环节依次走完Spark 里最常用的二十几个方法基本都能覆盖到后续再学窗口函数、Join、Streaming 时也不会觉得基础是悬空的。1.2 为什么选 PySpark 加 Spark SQL 的方式而不是纯 RDD 练习真的建议新手从 PySpark 的 DataFrame 和 Spark SQL 入手而不是把纯 RDD 算子当作主练习。原因很简单RDD 的很多底层算子实际开发中用得越来越少而 DataFrame 背后的 Catalyst 优化器能自动做谓词下推、列裁剪等优化这才是日常处理结构化数据的核心方式。用 PySpark 写练习还有一个实际优势你在交互环境里写 Scala每按一次回车等着编译切到 Python 后可以快速df.show()看结果、改逻辑、再运行试错成本低很多。有些人总觉得“生产环境用 Scala所以练习也要 Scala”这个说法至少在这个阶段不成立。你用 Scala 还是 Python只要最终生成的是同样的 DataFrame 和 SQL 任务物理执行计划不会有本质差别。练习的核心应该放在算子的语义和数据的流转上而不是被语言语法干扰。等你以后真去改 Spark 源码或者做底层性能优化再学 Scala 也来得及。2. 五分钟搭好本地环境并生成一份模拟订单数据2.1 本地环境搭建JDK、虚拟环境和 pip 安装的省心组合本地跑 PySpark 其实比想象中简单。我先说版本组合我用的 Spark 3.5.x配合 Python 3.9JDK 用 17。你用 JDK 8 或 11 也行但不建议直接用最新的 JDK 21因为部分 Spark 版本还不支持。安装步骤可以压缩成三步先确保java -version能输出版本号再创建一个 Python 虚拟环境最后在虚拟环境里用 pip 安装 PySpark。python -m venv spark_practice source spark_practice/bin/activate # Windows 下是 spark_practice\Scripts\activate pip install pyspark3.5.1 pandas这里我特别想提醒一句一定要固定版本。我见过有人直接pip install pyspark装到最新版结果跟着旧教程写代码API 对不上最后一堆报错。指定版本号后至少你能确定当前代码所对应的行为。还有一点pip 安装的 PySpark 会自带本地运行所需的基础组件你不需要单独装一份 Hadoop启动local[*]模式时会直接以本地多线程方式模拟分布式执行。2.2 用脚本生成五万行订单 CSV让练习有真实感不要随便在网上找一份不知道格式的数据来练最好自己用脚本生成一份可控制的模拟订单数据。字段设计要有电商业务的真实感但又要方便后续聚合订单号、用户ID、下单日期、商品类目、金额、区域、订单状态。我生成五万行日期分布在一到四月份金额在 10 到 5000 之间状态里有已完成、已取消、已退款。这样的数据既能做过滤也能做按月的分组统计。下面这个脚本可以直接保存成generate_data.py运行import random import datetime import csv random.seed(42) categories [手机数码, 家用电器, 服饰鞋包, 食品生鲜, 美妆个护] regions [华东, 华北, 华南, 西南, 东北] status_list [已完成, 已取消, 已退款] with open(data/sales.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([order_id, user_id, order_date, category, amount, region, status]) for i in range(50000): days random.randint(0, 120) order_date datetime.date(2024, 1, 1) datetime.timedelta(daysdays) writer.writerow([ fO{i:06d}, fU{random.randint(1, 2000):05d}, order_date.isoformat(), random.choice(categories), round(random.uniform(10, 5000), 2), random.choice(regions), random.choice(status_list) ])写文件时千万记得带换行处理参数和 UTF-8 编码。如果你在 Windows 上不小心用了系统默认编码后面 Spark 读出来中文可能直接乱码。生成之后可以先看前几行确认是正常 CVS 结构再进入下一步。3. 核心实操从读 CSV 到出报表的完整链路3.1 初始化 SparkSession把 CSV 变成 DataFrame在 PySpark 里所有功能的入口都是SparkSession。小练习不需要花里胡哨的配置我通常只设置三个东西应用名、运行模式、shuffle 分区数。运行模式写成local[*]表示使用本机所有可用线程spark.sql.shuffle.partitions设置为 8是为了避免默认 200 个 shuffle 分区在小数据量下造成大量空任务。from pyspark.sql import SparkSession spark (SparkSession.builder .appName(spark-small-practice) .master(local[*]) .config(spark.sql.shuffle.partitions, 8) .getOrCreate()) df spark.read.csv(data/sales.csv, headerTrue, inferSchemaTrue) df.printSchema() df.show(5, truncateFalse)inferSchemaTrue会让 Spark 自动推断每列类型对练习很方便。不过生产环境不要依赖这个参数一是它需要额外扫描一遍数据二是推断结果可能和你预期不一致。更好的做法是像下面这样显式定义 Schema尤其在实际团队协作时数据类型的稳定性比代码少几行重要得多。from pyspark.sql.types import (StructType, StructField, StringType, IntegerType, DoubleType, DateType) sales_schema StructType([ StructField(order_id, StringType()), StructField(user_id, StringType()), StructField(order_date, StringType()), StructField(category, StringType()), StructField(amount, DoubleType()), StructField(region, StringType()), StructField(status, StringType()) ])这一步做完df就是一张“可以横着拆开、竖着切分”的表。建议你花两分钟把printSchema的输出抄在笔记里后面写代码时时刻对照字段类型能省掉不少因类型误用导致的报错。3.2 数据清洗先过滤脏数据和重复订单很多人一拿到 DataFrame 就急着groupBy这是练习里最容易踩的坑。真实的订单数据一定有取消单可能还有重复入库的脏数据如果你不过滤掉聚合结果就会和业务口径对不上。我在这份模拟数据里只统计“已完成”的订单同时按order_id去重再排除金额为空或金额小于等于 0 的记录。日期字段读取后是字符串所以还得先转成真正的日期类型再拆出年份、月份和星期。from pyspark.sql import functions as F df_clean (df .filter(df[status] 已完成) .dropDuplicates([order_id]) .withColumn(order_date, F.to_date(order_date, yyyy-MM-dd)) .withColumn(year, F.year(order_date)) .withColumn(month, F.month(order_date)) .withColumn(week_of_year, F.weekofyear(order_date)) .filter(F.col(amount).isNotNull() (F.col(amount) 0))) print(清洗前记录数, df.count()) print(清洗后记录数, df_clean.count())你可以用这个count对比直观感受一下什么叫“数据质量影响结果”。在这个小练习里过滤前后甚至能差出几千条订单如果直接把脏数据也算进销售额最后报表一定会被挑战。关于withColumn我想多说一句它是生成新列不会修改原来的DataFrame所以链式调用时每步都必须接收返回值。我看到有初学者写df.withColumn(...)后以为原表变了下一行继续用df结果新列永远不生效。这是 PySpark 和 Pandas 在语义上一个很大的区别。3.3 分组聚合DataFrame API 和 Spark SQL 各写一遍清洗完成之后就到了练习的重点计算每个月份、每个品类的订单量、总销售额和平均客单价。我建议你用两种方式各写一遍先看 DataFrame 写法再看 Spark SQL 写法。DataFrame 写法如下result_df (df_clean .groupBy(category, month) .agg( F.count(order_id).alias(order_cnt), F.round(F.sum(amount), 2).alias(total_amount), F.round(F.avg(amount), 2).alias(avg_amount) ) .orderBy(month, F.desc(total_amount))) result_df.show(30, truncateFalse)这里groupBy里的列对应 SQL 的GROUP BYagg里的每一行则是一个聚合表达式。alias用来给输出列起名别小看这个动作生产报表里下游经常靠列名对接名字混乱很麻烦。Spark SQL 写法则是先把DataFrame注册成临时视图再直接写标准 SQLdf_clean.createOrReplaceTempView(sales) sql_result spark.sql( SELECT category, month, COUNT(order_id) AS order_cnt, ROUND(SUM(amount), 2) AS total_amount, ROUND(AVG(amount), 2) AS avg_amount FROM sales GROUP BY category, month ORDER BY month, total_amount DESC ) sql_result.show(30, truncateFalse)很多同学会问既然结果一样练一种不就好了我的看法是这两种写法你要至少各练一遍因为工作里一半的人用 DataFrame API另一半写 SQL你看不懂任何一方都会很吃力。更关键的是你得在脑子里建立起两者的对应关系SELECT对应selectWHERE对应filterGROUP BY对应groupByORDER BY对应orderBy。建立不了这个映射你读别人代码就像看天书。写完后可以用下面的方式快速验证两个结果是否一致避免自己骗自己diff_count result_df.exceptAll(sql_result).count() print(两张结果表的差异行数, diff_count)3.4 写出结果并回读校验确保不是“看起来对”Spark 写 CSV 和 Pandas 写 CSV 有一个很不一样的地方Spark 写出的不是单个文件而是一个目录里面会有多个part-*文件外加一个空白的_SUCCESS标记文件。这个设计是为了并行写出但在小练习里我们通常希望最终能看到一个可读的单文件。所以我会在写出前用repartition(1)把结果合并到一个分区output_path output/sales_summary sql_result.repartition(1).write.mode(overwrite).option(header, True).csv(output_path)这里要说明一下repartition(1)是重新落一次全量数据数据量大的时候会造成后续任务单分区处理、非常慢甚至 OOM练习里数据量小无所谓千万别把它当成生产环境写单文件的通用手段。如果生产上确实需要单文件一般先正常写出再单独做文件合并。写完以后不要直接打开 CSV 看一眼就说完成更严谨的做法是用 Spark 把输出再读回来和内存里的结果做对比check_df spark.read.csv(output_path, headerTrue, inferSchemaTrue) print(输出文件记录数, check_df.count()) check_df.show(10, truncateFalse)这样能发现很多隐藏问题。比如字段顺序变了、数值被截断、分区文件太多都可以在回读阶段暴露。我自己的习惯是任何一次作业写到文件系统之后都要立刻读一遍哪怕只是打印前二十行也不要跳过去。3.5 用 explain 和 Spark UI 验证执行细节跑通结果只是第一步我并不建议你停在“能出数”就结束。如果身边有人带你他大概率会问你一句你这个任务到底分成了几个 Stage有没有 Shuffle要回答这个问题最简单的方式是看执行计划。在 PySpark 里可以对一个 DataFrame 调用explainsql_result.explain(modeextended)不要被一长串输出吓到你只需要关注几个点扫描数据源有没有做列裁剪和谓词下推聚合过程中有没有出现Exchange节点也就是 Shuffle最后的结果有没有走排序。对这个小练习来说GROUP BY一定会触发 Shuffle所以你会看到Exchange这是正常的。如果还想看更直观的进度运行作业时打开浏览器访问http://localhost:4040也就是 Spark UI。你能看到 Job、Stage、Executor 的基本情况。本地练习虽然远没有集群环境那么壮观但尽早培养看 UI 的习惯对以后排查慢任务、数据倾斜会非常有帮助。4. 小练习高频问题排查与避坑技巧4.1 报错速查本地环境最容易踩的五个坑做练习的时候最打击信心的事情就是代码还没跑起来先被环境折腾半天。我把自己踩过和帮别人看过的五个高频报错整理在下面。第一个是启动 Python 时出现Unsupported class file major version这基本可以断定是 JDK 版本和 Spark 不匹配。我的处理办法是下载 JDK 17 并重新设置JAVA_HOME确保java -version输出为 17。第二个是 Windows 下报Could not locate executable null\bin\winutils.exe这是本地缺少 Hadoop 运行时导致的解决方法要么补齐对应的依赖文件要么干脆用 WSL 或 Linux 环境。我自己的经验是为一个小练习去折腾本地 Hadoop 相关组件不太划算能换环境就换环境。第三个问题很常见但经常被忽视spark.read.csv读不到文件。有人会把路径写成带盘符的绝对路径这在本地没问题但如果你的代码里包含中文路径或特殊字符就会有一堆编码问题。我更推荐把练习文件放在项目目录下用相对路径并确认启动 Python 的工作目录在项目根目录。第四个问题是写 CSV 后中文乱码一般发生在默认编码不是 UTF-8 的场景。你可以在写出时加option(encoding, UTF-8)生成数据时也保证用 UTF-8 编码。最后一个坑不是报错而是结果对不上明明按月份分组但输出的月份顺序是乱序的因为 Spark 默认不保证排序你需要显式排序。4.2 一个常见误区本地数据量小为什么还是慢我见过有人拿着这份小练习跑完一看发现很多 Stage 只有几百毫秒但总时间却要十几秒于是怀疑 Spark 是不是“太重了”。其实这多半不是 Spark 的问题而是你没有控制任务粒度。默认情况下Spark SQL 的spark.sql.shuffle.partitions是 200意思是任何 Shuffle 操作都会分成 200 个分区去跑。你的数据才五万行分成 200 个分区只会产生大量空任务和调度开销。我在练习开始时把它改成了 8整个聚合任务明显轻快不少。还有一个容易被忽略的点count()是一个 Action每次调用都会从源头重新计算一遍。比如你在代码里写了df.count()又写了df_clean.count()再运行一次聚合前面几步可能重复执行了多次。如果数据量级稍微大一点这种重复成本会被放大。你可以用df.cache()把中间结果缓存起来不过缓存本身也有代价要放在多次复用的场景下才划算。小练习里最值得学的不是堆缓存而是搞清楚哪些算子会触发 Shuffle、哪些操作是 Action。你只要把这两点分清就不难理解“慢”到底慢在哪个环节。4.3 经典问题对照表现象、原因、处理方案每次带新人做这个练习我都会让他们对照一张问题表排查效率比自己瞎猜高很多。这里整理成表格放在你旁边对照即可现象常见原因处理建议启动报Unsupported class file major versionJDK 版本超出 Spark 支持范围使用 JDK 8/11/17重新设置 JAVA_HOMEWindows 下报 winutils.exe 找不到缺少 Hadoop Windows 本地依赖补齐依赖或改用 WSL/Linux 环境中文读取或写出乱码文件编码不是 UTF-8读写时统一指定encoding文件本身也用 UTF-8 保存结果月份顺序乱没有显式排序或排序字段类型是字符串对日期字段转成 date 类型后再排序或加ORDER BY month数据特别小却跑很慢默认 shuffle 分区数过大产生大量空任务设置spark.sql.shuffle.partitions为较小值聚合值比预期大没过滤取消单、没去重、空值未处理先做数据质量检查再进入汇总这张表只解决“能不能跑”的问题真正理解为什么还需要你回到前面的章节把每条现象背后的 Spark 原理串起来。比如为什么分区数影响速度为什么去重要放在聚合之前这些不是靠背而是靠亲手改参数、观察结果差异来获得体感。5. 做完练习之后的几条实在体会5.1 跑通只是第一步重点是看懂每一条转换把这组练习完整跑完之后我最想说的一点是不要满足于“代码能出结果”。如果你的操作只是照着贴一遍那你的提升非常有限。Spark 的学习曲线不是靠记忆弯道超车而是靠对数据流转的想象。每调用一个withColumn都会生成一条新的数据版本每执行一次groupBy数据都会发生一次宽依赖也就是 Shuffle。你要在脑子里形成一条“数据从文件出发经过哪几步变成什么形状最终落在哪里”的路径。做完练习后我建议你把每一行代码删掉再重写一遍不用看答案能写出来才算真会了。这个复写的过程比复制十遍都管用。5.2 同一个数据集还能怎么练出更多东西这个练习的数据集其实很有挖头做完基础版本之后你可以继续加难度。比如用窗口函数计算每个品类按月销售额的环比看哪些品类增长明显也可以用用户维度做一次关联查询把订单表和用户表关联后分析复购率还可以把月结果和累计结果放在一起试一试ROLLUP或CUBE。这些扩展仍然只用这一份模拟数据成本很低但每次都能把 Spark 的一个知识块补得更完整。我自己做这个练习时额外加了一个读取目录下多个 CSV 文件的小任务本质上和读取单文件一样但顺带理解了通配符路径和分区发现的机制。等你把这些都试完再回头看最初的小练习会发现它就是一个很好的 Spark 学习坐标系。如果对你有帮助可以按我上面的步骤自己跑一遍尤其是问题排查那部分踩过一次坑记住的细节比刷十道题都牢。我最后再多说一句本地 Spark 小练习只是起点后续接上真实数据源、加上合理的分区策略、学会看执行计划你会发现 Spark 那些看起来吓人的概念其实都能在这次练习里找到落点。