外卖订单2000万行单机跑不动？用Spark做分布式分析与可视化

简介这份资源是面向高校计算机专业学生的毕业设计与课程设计参考项目主题为基于Spark的外卖大数据平台分析系统适合已具备一定大数据基础、需要完成实战项目或课程作业的学习者。压缩包共40个文件约645KB以Scala源码为主体辅以Markdown说明文档、HSQL与SQL脚本、JSON与CSV样例数据、XML配置及Python脚本等覆盖从数据采集、清洗到分析建模的完整链路。项目围绕Spark核心知识点展开包括RDD转换与行动操作、DataFrame与Dataset的SQL查询、Spark SQL多数据源交互、Spark Streaming实时流处理以及MLlib机器学习算法应用并涉及Driver、Executor与Worker的架构理解与性能调优思路。目前已有168人学习下载可作为毕业设计选题落地的完整参考帮助读者掌握大数据平台的设计流程与工程实现方法。1. 外卖订单数据堆到 2000 万行之后单机 pandas 为什么开始集体翻车去年帮一个做计算机课程毕设的学弟看进度他那个「基于 Spark 的外卖大数据平台分析系统」卡在了最尴尬的位置数据量刚过 2000 万行订单单机 pandas 跑一次全量聚合要四十多分钟中途内存直接爆掉Jupyter 内核重启了三次。他一开始以为是代码写得烂把groupby改成了pivot_table又换成numpy手写循环结果只是把崩溃时间从第 12 分钟推迟到第 18 分钟。问题根本不在写法而在单机内存模型——pandas 默认要把整列读进内存再算2000 万行订单加上用户、商家、配送三张维表做 join峰值内存轻松突破 16GB普通笔记本根本扛不住。这个标题讲的就是把这类外卖业务数据从「单机跑不动」搬到 Spark 上做分布式分析订单量、客单价、时段分布、区域热力、商家排名、配送时长这些指标用 Spark SQL 和 DataFrame 重新实现一遍再配一个能看结果的可视化层。它适合两类人一类是正在做大数据方向毕设、需要一套能跑通、能答辩、能写进论文的系统另一类是刚接触 Spark、想找一个真实业务场景练手的工程师。核心不是把 Spark 当玩具跑个 WordCount而是让它真正处理一份有业务含义、有脏数据、有倾斜问题的订单数据集。2. 先把数据流和分层想清楚ODS、DWD、DWS 三层怎么切2.1 为什么外卖场景必须做分层不能一张宽表算到底外卖订单的原始数据通常来自两个方向一是业务库导出的订单明细字段包括订单 ID、用户 ID、商家 ID、下单时间、支付时间、配送时间、订单金额、配送费、优惠金额、订单状态二是维表用户表、商家表、区域表、骑手表。如果直接把订单明细和所有维表 join 成一张大宽表再算指标会踩两个坑第一维表更新后历史宽表要重算成本极高第二不同指标需要的粒度不一样客单价按订单算区域热力按区域算骑手效率按骑手算一张宽表满足不了所有粒度。常见做法是分三层。ODS 层保留原始数据不做任何清洗只做分区落地按天分区。DWD 层做清洗和标准化过滤掉取消订单、异常金额订单把时间字段统一成标准格式把用户、商家、区域维表打平进来形成订单粒度的事实表。DWS 层按主题聚合按天、按区域、按商家、按小时分别汇总每个主题一张表。这样上层做报表或者可视化时直接查 DWS 层不用每次扫全量明细。我一般会建议毕设项目至少把 DWD 和 DWS 两层做出来ODS 层可以用一份模拟数据文件代替不必真的接 Kafka。因为答辩老师更关心你有没有分层意识、有没有处理数据倾斜、指标口径是否一致而不是你有没有搭一套完整的采集链路。2.2 用 Spark SQL 建 DWD 订单事实表的最小代码下面这段代码是在本地模式跑通的 DWD 层构建逻辑读的是 CSV 格式的模拟订单数据输出 Parquet 分区表。字段名按外卖业务常见口径命名实际替换成你自己的列名即可。from pyspark.sql import SparkSession from pyspark.sql.functions import col, to_timestamp, when, datediff, unix_timestamp # 本地模式启动driver 内存给到 4G避免小数据集也 OOM spark SparkSession.builder \ .appName(waimai_dwd_build) \ .master(local[4]) \ .config(spark.driver.memory, 4g) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate() # 读原始订单 CSV第一行是表头自动推断类型 raw_order spark.read \ .option(header, true) \ .option(inferSchema, true) \ .csv(data/ods/order_raw.csv) # 读商家维表和区域维表 dim_shop spark.read.option(header, true).csv(data/ods/shop_dim.csv) dim_area spark.read.option(header, true).csv(data/ods/area_dim.csv) # 清洗过滤取消订单、金额为负、下单时间为空的记录 clean_order raw_order.filter( (col(order_status) ! cancelled) (col(order_amount) 0) (col(order_time).isNotNull()) ) # 时间标准化把字符串转成 timestamp并派生日期分区字段 clean_order clean_order.withColumn( order_ts, to_timestamp(col(order_time), yyyy-MM-dd HH:mm:ss) ).withColumn(dt, col(dt).cast(string)) # 打平维表商家所属区域、区域名称 dwd_order clean_order \ .join(dim_shop, onshop_id, howleft) \ .join(dim_area, onarea_id, howleft) \ .select( order_id, user_id, shop_id, area_id, area_name, order_ts, dt, order_amount, delivery_fee, discount_amount, pay_time, deliver_time, order_status ) # 按天分区写入 Parquet分区字段放最后 dwd_order.write \ .mode(overwrite) \ .partitionBy(dt) \ .parquet(data/dwd/dwd_order)这段代码的关键点有三个。第一spark.sql.shuffle.partitions在本地模式下调成 8默认 200 会让小数据集产生大量空 task反而拖慢速度。第二清洗条件里order_amount 0和order_status ! cancelled是最基本的业务过滤实际项目里还要处理重复订单、测试订单、金额异常大的刷单订单。第三维表 join 用 left join 而不是 inner join因为有些订单的商家可能已经下线维表里查不到inner join 会直接丢掉这些订单导致统计口径偏小。2.3 DWS 层聚合客单价、时段分布、区域热力三张主题表DWD 建好之后DWS 层就是按不同维度做 group by。下面这段代码一次性产出三张主题表分别对应客单价、时段分布、区域热力。from pyspark.sql.functions import sum as _sum, count, avg, hour, round as _round dwd spark.read.parquet(data/dwd/dwd_order) # 主题一按天、按区域的客单价和订单量 dws_area_daily dwd.groupBy(dt, area_id, area_name).agg( count(order_id).alias(order_cnt), _round(_sum(order_amount) / count(order_id), 2).alias(avg_order_amount), _round(_sum(delivery_fee), 2).alias(total_delivery_fee) ) # 主题二按小时的下单量分布用于看午高峰和晚高峰 dws_hourly dwd.withColumn(order_hour, hour(order_ts)) \ .groupBy(dt, order_hour) \ .agg(count(order_id).alias(order_cnt)) # 主题三按商家的订单量和销售额排名基础表 dws_shop_daily dwd.groupBy(dt, shop_id).agg( count(order_id).alias(order_cnt), _round(_sum(order_amount), 2).alias(total_amount) ) dws_area_daily.write.mode(overwrite).parquet(data/dws/dws_area_daily) dws_hourly.write.mode(overwrite).parquet(data/dws/dws_hourly) dws_shop_daily.write.mode(overwrite).parquet(data/dws/dws_shop_daily)客单价用sum(order_amount) / count(order_id)而不是avg(order_amount)是因为订单金额字段里可能包含配送费和优惠直接 avg 会把配送费也算进客单价口径不对。时段分布用hour()函数从 timestamp 里取小时注意这里取的是下单时间不是支付时间两者在业务上含义不同。商家排名表只存基础聚合结果具体排名在应用层用窗口函数或者排序取 TopN不在这里写死。3. 把 Spark 分析结果接到可视化层三种落地方式怎么选3.1 毕设场景下最稳的方案Spark 产出结果表Web 层直读毕设项目最容易翻车的地方不是 Spark 算不出来而是算完之后不知道怎么展示。常见做法有三种第一种是 Spark 把结果写到 MySQL 或者 Hive然后 Java 或 Python Web 后端查库返回 JSON前端用 ECharts 画图第二种是 Spark 直接输出 CSV前端用静态文件加载第三种是接 BI 工具。我一般会推荐第一种因为答辩时老师会问「你的系统架构是什么」你说 Spark 算完写 MySQLWeb 层查库展示这是一个完整的数据流听起来也合理。第二种太简陋第三种对毕设来说太重。具体做法是在 DWS 层算完之后用 Spark 的 JDBC 写入 MySQL。# 把 DWS 结果写入 MySQL供 Web 后端查询 dws_area_daily.write \ .format(jdbc) \ .option(url, jdbc:mysql://localhost:3306/waimai_analysis) \ .option(dbtable, dws_area_daily) \ .option(user, root) \ .option(password, your_password) \ .option(driver, com.mysql.cj.jdbc.Driver) \ .mode(overwrite) \ .save()这里要注意 MySQL 表的字符集要设成 utf8mb4否则区域名称里的中文可能乱码。另外mode(overwrite)会先 drop 再建表如果表结构已经手动建好并且有索引改成append更安全。写入前最好在 MySQL 里先建好表字段类型和 Spark 的 schema 对齐避免自动建表时类型映射出错。3.2 可视化层选 ECharts 还是 Superset看你的时间预算如果毕设时间充裕想做得好看一点可以用 Superset 或者 Metabase 直接连 MySQL拖拽生成图表省去写前端的时间。但这两个工具对新手来说部署成本不低Docker 拉起来之后还要配数据源、配 dashboard遇到端口冲突或者权限问题容易卡住。如果时间紧直接用 ECharts 写几个静态页面后端用 Flask 或 Spring Boot 提供三个接口区域热力、时段分布、商家 TopN前端 ajax 请求后渲染两天就能搞定。我见过一个比较聪明的做法是Spark 算完的结果表导出成 JSON 文件前端直接读本地 JSON 渲染完全跳过 Web 后端。这样架构上少一层答辩时如果老师问「为什么不用数据库」你可以说「结果表数据量小静态化可以降低系统复杂度」也算一个合理的解释。但缺点是数据更新要重新导出不适合实时场景。3.3 指标口径对齐三个最容易和答辩老师扯皮的地方第一个是「订单量」到底算不算取消订单。如果你的 DWD 层已经过滤了取消订单那 DWS 层的订单量就是有效订单量答辩时要主动说明。第二个是「客单价」是否包含配送费。外卖行业通常客单价指商品金额不含配送费但有些平台会把配送费算进去。第三个是「区域」是按用户收货地址还是按商家地址。这两个口径算出来的区域热力完全不同一个反映消费需求分布一个反映供给分布。建议在论文里单独用一小节说明指标定义避免答辩时被追问。4. 避坑与排查Spark 跑外卖数据最常见的五个翻车现场4.1 数据倾斜某个热门商家把 task 拖到超时现象是 Spark 任务跑到 99% 卡住不动打开 Spark UI 看 stage 详情发现某个 task 处理的数据量是其他 task 的几十倍。原因是 group by shop_id 时某个头部商家的订单量远超其他商家所有这个商家的记录都分到同一个 partition。解决办法有两个一是加盐给 shop_id 拼接随机后缀打散聚合两次二是如果只是 TopN 排名先用sortWithinPartitions加limit做局部 TopN再全局合并。毕设数据量不大时直接把spark.sql.shuffle.partitions调大也能缓解但治标不治本。4.2 时间字段解析失败导致大量 null现象是 DWD 层建完之后按小时聚合的结果全是 null。原因是原始 CSV 里的时间格式不统一有的是2024-01-01 12:00:00有的是2024/01/01 12:00to_timestamp用固定格式解析时对不上的直接返回 null。解决办法是先用date_format或者正则把时间字符串标准化再解析。或者用to_timestamp的多种格式尝试失败时记录到脏数据表里单独排查。4.3 小文件过多拖慢后续查询现象是 DWS 层写完之后用spark.read.parquet读回来发现分区目录下有几万个几百字节的小文件查询时启动 task 的时间比计算时间还长。原因是 shuffle 分区数设得太大每个分区写一个文件。解决办法是在写入前用coalesce或repartition控制文件数一般每个分区目录下 1 到 3 个文件比较合适。本地模式跑毕设数据coalesce(1)就够了。4.4 JDBC 写入 MySQL 时中文乱码现象是 Spark 写入 MySQL 后区域名称字段显示成问号或者乱码。原因是 JDBC URL 没指定字符集或者 MySQL 表默认字符集是 latin1。解决办法是在 URL 后面加?useUnicodetruecharacterEncodingutf8并且建表时指定DEFAULT CHARSETutf8mb4。如果已经写入乱码需要先 truncate 表再重新写直接改字符集不会修复已有数据。4.5 本地模式内存不够driver OOM现象是任务刚启动就报java.lang.OutOfMemoryError: Java heap space。原因是spark.driver.memory默认只有 1G读大 CSV 或者做 join 时不够用。解决办法是在SparkSession.builder里显式设置config(spark.driver.memory, 4g)或更高。注意这个参数在代码里设置和在spark-submit命令行设置效果一样但本地模式跑 Jupyter 时只能在代码里设。如果调到 8G 还 OOM说明数据量真的超过了单机处理能力需要考虑换集群或者采样。5. 从能跑到能答辩三个让系统看起来更专业的进阶技巧5.1 用窗口函数做商家排名而不是先全量排序再取 TopN很多人写商家排名时习惯orderBy(desc(total_amount)).limit(10)这在数据量大时会把所有数据 shuffle 到一个节点排序效率极低。更好的做法是用窗口函数在每个区域内先排名再过滤。from pyspark.sql.window import Window from pyspark.sql.functions import row_number, desc # 按区域分区按销售额降序排名取每个区域前 10 window_spec Window.partitionBy(area_id).orderBy(desc(total_amount)) shop_rank dws_shop_daily \ .withColumn(rank, row_number().over(window_spec)) \ .filter(col(rank) 10)这样每个区域并行排序不会把所有数据压到一个节点。窗口函数的partitionBy决定了并行度orderBy决定了排名规则。如果要做全局 TopN可以把partitionBy去掉但那样又回到单点排序的问题所以全局 TopN 更适合用sortWithinPartitions加limit的两阶段做法。5.2 加一层结果校验证明你的数字不是拍脑袋出来的答辩时老师经常会问「你这个订单量总数对不对」。如果你能现场跑一个校验脚本对比 DWD 层订单数和 ODS 层过滤后的订单数是否一致会显得很专业。校验逻辑很简单ODS 层总行数减去取消订单数减去异常金额订单数应该等于 DWD 层行数。如果不等说明清洗条件写错了或者 join 丢了数据。ods_cnt raw_order.count() cancel_cnt raw_order.filter(col(order_status) cancelled).count() abnormal_cnt raw_order.filter(col(order_amount) 0).count() dwd_cnt dwd_order.count() expected ods_cnt - cancel_cnt - abnormal_cnt assert dwd_cnt expected, f数据校验失败: DWD{dwd_cnt}, 预期{expected}这个断言跑通之后你可以在论文里写「系统包含数据质量校验环节确保分层之间数据一致性」比空泛地说「数据清洗」有说服力得多。5.3 把 Spark UI 的截图放进论文比文字描述更有力Spark 任务跑起来之后4040 端口有 Web UI能看到 stage、task、shuffle 读写、执行计划。答辩前把 DAG 图、stage 详情、SQL 执行计划各截一张图放进论文的「系统实现」章节配上一句「通过 Spark UI 观察任务执行情况确认无数据倾斜」比写一大段文字描述架构有效得多。如果本地模式跑不出来 UI可以在spark-submit时加--conf spark.ui.port4040显式指定端口。我自己做这类项目最大的教训是不要一上来就追求「全量数据 完整链路」先用 10 万行采样数据把 ODS 到 DWS 到 MySQL 到前端的链路跑通再逐步加数据量。很多毕设卡住不是因为技术难而是因为一开始就把目标定得太大结果每一步都在调环境、调内存、调依赖最后时间全花在排错上核心逻辑反而没写多少。先把最小闭环跑通再往上加指标、加维度、加可视化这才是能按时交付的做法。希望帮到你。本文还有配套的精品资源点击获取