双11场景下Spark离线数据分析实战：从数仓分层到数据倾斜调优

双11差不多是国内电商每年最大的一场技术大考。流量洪峰、订单暴涨、大屏滚动，背后全是数据处理的活儿。这篇文章整理的是我在某电商平台双11大数据分析项目中负责的Spark离线分析部分，覆盖指标口径、数仓分层、核心分析代码、数据倾斜调优、资源治理和结果落库，基本是把“拿到一堆原始日志，到最终产出可看的数据报表”这条链路完整走了一遍。适合正在做大数据分析项目、想了解Spark实践细节的同学，也适合做离线数仓的朋友拿来参考。

1. 双11分析项目到底在算什么

1.1 业务场景倒推技术需求

双11大促期间，用户行为日志和订单量级会迎来一年中最高的峰值。单日产生的用户行为记录（浏览、收藏、加购、下单、支付）通常在百亿条以上，订单明细也是亿级。如果想回答“今天卖了多少”“哪个品类最猛”“哪个地区购买力最强”“用户是从哪个环节流失的”，靠常规的数据库查询方式基本跑不出来，这时就需要Spark这类分布式计算引擎做离线的批量处理。

你可以把Spark在这里的角色理解成一条自动化的数据加工流水线：源头是源源不断进来的原始日志，输出是结构清晰的分析结果表，中间是清洗、关联、聚合、排序这一系列操作。相比Hive MapReduce，Spark基于内存计算的执行模型在跑这类多阶段复杂作业时快得多；相比Flink实时处理，它在处理动辄百亿级的历史全量数据和复杂关联分析时成本更低、逻辑更稳。双11这种场景往往是流批两条腿走路，实时链路负责大屏分钟级刷新，离线链路负责全量精确计算和深层分析，我们这个项目就是离线这半边。

1.2 指标清单和分析口径必须提前锁死

项目启动后，第一件事不是写代码，而是和业务方对齐口径。这块如果先动工，后面返工成本极大。举几个我们当时拉出来的核心指标清单，你可以直观感受一下：

指标口径要注意的点其实很细：比如用户数到底按什么ID去重，是登录用户还是设备ID；访问UV是否包含爬虫流量；加购转化率的分子分母时间窗口怎么对齐；预售定金算不算GMV。我们当时的做法是每项指标都写进一份口径文档，用具体例子解释清楚，然后让业务方签字确认。宁可前期多花两天扯皮，也好过数跑了三版对不上数。

2. 数据准备：从原始日志到可分析的宽表

2.1 ODS层清洗，垃圾数据不过夜

分析代码写得再快，数据源脏乱差也白搭。原始日志进入Spark分析前，必须经过ODS（原始数据层）的清洗加工。

我们每天上千亿条日志跑的是离线ETL批作业，凌晨把前一天所有的日志从消息队列或者业务库同步到HDFS，然后交给Spark程序处理。清洗这件事，看起来就是“过滤脏数据”，但做起来都是细节：

val odsBehaviorDF = spark.read.json(rawPath) .filter(col("event_time").isNotNull) .filter(col("dt") === bizDate) .filter(col("user_id").isNotNull && length(col("user_id")) > 0) .filter(col("item_id").isNotNull) .filter(col("behavior_type").isIn("view", "fav", "cart", "buy"))

注意几个点：

• 过滤时间窗口外的数据。日志数据经常出现迟到或者时间错乱，比如前一天凌晨的数据，第二天上午才补传上来。我们要按业务时间过滤，而不是按入库时间。

• 识别重复上报。客户端会在异常情况下重试上报，同一个事件会出现多条记录。早期我们没处理，发现统计出的UV和业务方线上系统对不上。后来加了event_id字段做去重，即使重复上报也能保证同一事件只计入一次。

• 爬虫和机器流量识别。大促期间爬虫流量尤其多，会影响PV、UV这些指标。我们维护了一个UA黑名单和IP特征库，清洗阶段直接剔除典型的机器流量。业务方自己也要在报表层再做一次过滤，两层保险。

2.2 DWD层明细宽表，一次建模，处处复用

ODS层的数据还是分开的：行为日志在A表，订单信息在B表，商品维度在C表。如果每个分析任务都直接对这些原始表做join，既慢又乱，而且不同开发写出来的join逻辑可能不一致。

所以接下来要做DWD层（明细数据层）的宽表设计。核心思想非常简单：把最常用的维度通过一次join统一冗余到事实表上，形成“一张大宽表”。

以交易分析为例，我们把订单事实表和商品维表、类目维表、店铺维表做关联，加工成一张dwd_trade_order_wide表，字段至少包含：

• 订单维度：订单ID、下单时间、支付时间、订单状态、支付金额、优惠金额
• 用户维度：用户ID、用户注册时间、用户等级、会员标识
• 商品维度：商品ID、商品名称、品牌、一级类目、二级类目、店铺ID
• 地域维度：收货省份、收货城市

行为分析链路同理，把用户行为日志和商品维度冗余，得到dwd_behavior_wide表。

这样做的好处是：下游所有分析任务只需要从这一张宽表跑，不需要反复join。对双11这种“任务多、时间紧、资源贵”的攻防期场景来说，宽表牺牲了一点存储，换来了大量的计算时间和开发效率，非常划算。

2.3 存储格式与分区策略决定查询下限

数据从ODS到DWD，存储上我们统一用Parquet + Snappy压缩。Parquet是列式存储格式，聚合分析时只需要读取涉及的那几列，能省大量I/O；配合谓词下推，即使数据量很大，过滤条件也能在读取阶段提前裁掉不需要的数据。

分区策略也是踩过坑才有结论的。对于订单表，按日分区是最基本的，但双11、双12这种流量暴涨的日期，一天的数据量比其他月份一整个月还多，单日分区压力太大。我们在日分区下面再加了小时分区，比如dt=2024-11-11/hour=00，既方便跑小时级快照，也能让某些修复任务只重跑特定小时段，省资源。

这里有个常用技巧，建表时加上适当的文件合并逻辑。Spark写Parquet文件时很容易产生大量小文件，导致后续读取任务被文件打开开销拖垮。我们每执行完一次ETL，都会跑一遍OPTIMIZE或者用repartition控制输出文件数量，保证每个输出文件在256MB左右。

3. 核心分析功能的Spark实现细节

3.1 大盘指标聚合，先懂逻辑再写代码

大促当天最核心的数据一定是“实时大屏”，但离线侧也不会闲着，第二天凌晨要产出全量精确结果，作为各种复盘报表的基础。GMV、订单量、支付用户数这些指标，在Spark里做起来门槛不高，就是分组聚合。

SELECT COUNT(DISTINCT user_id) AS pay_uv, COUNT(order_id) AS order_cnt, SUM(pay_amount) AS gmv FROM dwd_trade_order_wide WHERE dt = '2024-11-11' AND pay_status = 'paid'

这里有个容易被忽略的细节：COUNT(DISTINCT ...)在数据量特别大时性能和结果精确度会有一些取舍。双11级别的用户数做精确去重，需要把分散在几百上千个分区的数据集中到一起，shuffle量巨大，耗时极高。

我们当时用了approx_count_distinct，它是基于HyperLogLog的实现，在数据量极大、指标精度允许一定误差的情况下（比如UV统计，误差率2%左右业务方通常可以接受），能在千亿级数据上秒级完成去重统计。使用时注意合理设置spark.sql.adaptive.enabled下的相关参数，能在结果准确度和性能之间找到一个平衡。

如果是严格对账用的指标，比如总订单量、总GMV，必须精确统计，不要用近似函数；但像“访问用户数”这种本身定义就有边界的数据，用近似统计收益远大于风险。

3.2 漏斗分析：每一步的流失藏在哪里

用户从进入页面到最终支付，中间会经历浏览、加购、下单、支付多个环节。分析团队很关心每个环节的转化率，哪个环节流失最多，哪个环节存在异常。

实现漏斗最方便的做法是用条件聚合。我们以用户为粒度，统计每个行为是否发生过，然后逐级计算转化：

SELECT COUNT(DISTINCT CASE WHEN view_cnt > 0 THEN user_id END) AS view_uv, COUNT(DISTINCT CASE WHEN cart_cnt > 0 THEN user_id END) AS cart_uv, COUNT(DISTINCT CASE WHEN order_cnt > 0 THEN user_id END) AS order_uv, COUNT(DISTINCT CASE WHEN pay_cnt > 0 THEN user_id END) AS pay_uv FROM ( SELECT user_id, SUM(CASE WHEN behavior_type = 'view' THEN 1 ELSE 0 END) AS view_cnt, SUM(CASE WHEN behavior_type = 'cart' THEN 1 ELSE 0 END) AS cart_cnt, SUM(CASE WHEN behavior_type = 'buy' THEN 1 ELSE 0 END) AS order_cnt FROM dwd_behavior_wide WHERE dt = '2024-11-11' GROUP BY user_id ) t

这段SQL的逻辑是：先按用户聚合出各行为类型的次数，再统计“发生过”的用户数。漏斗的每一级都是上一级的子集，最后用后一级除以前一级得到转化率。

实际中还要考虑“是否只看首次行为”。比如一个用户先浏览再下单，后又回来浏览，浏览和支付的先后顺序怎么算？我们当时定的口径是：以支付行为为终点倒推漏斗，只看支付前最近一次的行为链路。这就需要在子查询里用窗口函数做时间排序，复杂度更高一点，但更贴近真实业务。

3.3 窗口函数做复购偏好，比分组聚合灵活太多

用GROUP BY做出来的结果只是一行行的汇总数，但如果想看“每个用户购买次数最多的类目Top3”，或者“每个用户在双11前后15天的复购周期”，就需要窗口函数。

统计用户购买类目偏好，代码写出来很直观：

SELECT user_id, cat_level_1, cnt FROM ( SELECT user_id, cat_level_1, COUNT(DISTINCT order_id) AS cnt, ROW_NUMBER() OVER(PARTITION BY user_id ORDER BY COUNT(DISTINCT order_id) DESC) AS rn FROM dwd_trade_order_wide WHERE dt = '2024-11-11' GROUP BY user_id, cat_level_1 ) t WHERE rn <= 3

窗口函数最需要注意的坑是分区数据的处理。大促用户量大，普通的PARTITION BY user_id在几百亿数据上可能造成单个分区的数据量过大，产生数据倾斜。这种情况可以配合Salting key或者Session窗口进行二次拆分，后面调优章节会详细展开。

复购周期的计算也靠窗口函数。我们当时要算“用户两次购买之间的间隔天数”，直接对同一用户按购买时间排序，取前后两单的日期差值：

SELECT user_id, order_dt, LAG(order_dt, 1) OVER(PARTITION BY user_id ORDER BY order_dt) AS prev_order_dt, DATEDIFF(order_dt, LAG(order_dt, 1) OVER(PARTITION BY user_id ORDER BY order_dt)) AS gap_days FROM dwd_trade_order_wide WHERE pay_status = 'paid'

很多业务问题一旦落到“窗口内比较前后行”的思路上，用LAG、LEAD、ROW_NUMBER、RANK就能非常自然地解掉。

3.4 TopN排行：别把结果全堆到Driver端

大促结果出来后，运营团队最喜欢看的就是各种TopN：品类销售Top10、品牌成交Top20、单品爆款Top100、省份购买力Top10。Spark实现TopN看起来是order by + limit，但数据量大时有一个性能大坑：全量排序会把大量数据shuffle到同一个task，然后再取TopN，Executor内存压力非常大。

更稳的写法是分布式的两阶段取TopN。先用分区内排序取TopN，再做全局合并：

val topN = df .groupBy(col("cat_level_1")) .agg(sum("pay_amount").alias("gmv")) .repartition(col("cat_level_1")) .sortWithinPartitions(col("gmv").desc) .limit(20)

sortWithinPartitions只对每个分区内数据进行排序，不会触发全量全局排序，再配合limit算子，执行效率比orderBy高不少。我们线上全部TopN类作业都改成了这种写法，对比下来时间大约节省了六成。

4. 数据倾斜与大表关联：双11作业的噩梦现场

4.1 任务卡在99%的现场复盘

双11当天的数据天然具备倾斜特质：头部品牌、爆款商品、核心地域的流量远超普通数据。如果按这些维度做聚合或者关联大表，很快会看到Spark UI里出现一个诡异的现象：99%的task都跑完了，剩下那一个task处理时间比所有人都长，整个stage卡住不动。轻则作业多跑几分钟，重则OOM直接失败。

我们第一次碰到这种现象是在做商品维度关联时。订单表关联商品维表，理论上几亿订单散布在上百万商品上，问题是某几个爆款商品的订单量占了全表的30%，所有订单在shuffle时都涌向同一个分区，处理该分区的Executor成为瓶颈。

当时一看Spark UI，那个task的Shuffle Read Size是其他task的几十倍，磁盘溢写严重，GC时间超高，一眼就断定是数据倾斜。这玩意儿不查会一直跑失败，查了就逃不脱下面这些处理手法。

4.2 加盐、热点打散、广播的取舍与组合

对症下药要分清场景。

场景一：热点Key关联维表或事实表。

最有效的手段是加盐（Salting）。思路很简单：给热点Key加上随机前缀，把它拆成很多个子Key，让它们分散到不同分区去处理。比如热点商品ID=10086，可以把它改造成10086_0、10086_1、10086_2直到10086_9，然后关联时也把维表复制10份。

val saltedFact = factDF .withColumn("salt", lit(rand() * 10) cast IntegerType) .withColumn("join_key", when(col("item_id").isin(hotItems: _*), concat(col("item_id"), lit("_"), col("salt"))) .otherwise(col("item_id"))) val saltedDim = dimDF .withColumn("salt", explode(typedLit(0 to 9))) .withColumn("join_key", when(col("item_id").isin(hotItems: _*), concat(col("item_id"), lit("_"), col("salt"))) .otherwise(col("item_id")))

注意加盐后要把结果再按原始Key做一次groupBy，把分散的数据重新合并，否则统计结果会按拆分后的子Key分多行。

场景二：小表关联大表。

如果维表只有几十万行，订单表几十亿行，直接用Broadcast Join即可。把维表存入每个Executor的内存里，压根不会有shuffle。

我们当时把商品维表、类目维表、地域维表全做了广播处理，条件是spark.sql.autoBroadcastJoinThreshold设置为10MB——超过临界值则自动退化为SortMergeJoin。

场景三：倾斜无法定位热点，只能用skew join。

Spark 3之后开启AQE，自动处理一部分倾斜情况。参数打开即可：

spark.sql.adaptive.skewJoin.enabled=true spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB

它会自动检测Shuffle阶段哪些分区数据量过大，然后把它拆成多个分区再处理，减少单task压力。实测确实有效，但注意它只针对SortMergeJoin的场景有效，且需要开启spark.sql.adaptive.enabled=true。

4.3 AQE与动态合并的实际效果

AQE（Adaptive Query Execution）除了处理倾斜，还带来一个非常有用的能力：动态调整Shuffle分区数。

默认情况下，spark.sql.shuffle.partitions设置为200。如果数据量特别大，200个分区可能不够，导致每个task处理的数据过大；如果数据量小，200个分区又造成大量空任务，白白浪费调度开销。AQE可以通过spark.sql.adaptive.coalescePartitions.enabled=true在运行时自动合并过小的分区。我们开启后，双11当天一个常规聚合作业的task数量从200降到了48，整个作业耗时缩短了接近一半。

这一章节强烈建议所有跑大数据的同学认真对待。倾斜不是“偶发问题”，而是大促数据的必然。不要等生产事故了再临时头疼，上面的加盐、广播、AQE组合拳，在日常调度里就要沉淀成模板。

5. Spark资源参数与运行稳定性治理

5.1 Executor内存、并行度和数据量的匹配

分析代码写得再优美，跑在资源不合理的Spark集群上一样会挂。双11期间集群资源紧张，给每个作业分配到的资源是固定的，能做的就是让作业在有限资源下跑得最快。

我们采用的典型参数模板如下：

spark.executor.memory=8g spark.executor.memoryOverhead=2g spark.executor.cores=4 spark.dynamicAllocation.enabled=true spark.dynamicAllocation.minExecutors=10 spark.dynamicAllocation.maxExecutors=200 spark.sql.shuffle.partitions=400 spark.sql.adaptive.enabled=true spark.sql.adaptive.coalescePartitions.enabled=true spark.sql.adaptive.skewJoin.enabled=true spark.sql.autoBroadcastJoinThreshold=10485760

一个关键经验：Executor内存不要贪大。8g到16g是比较稳妥的范围，再大反而增加GC压力；配合executor.cores=4，能较好平衡并行度和单task资源。executor.memoryOverhead一定要设置，否则遇到大数据量shuffle时OOM风险极高，尤其使用Parquet读取时。

并行度的设置要理解分区的含义。spark.sql.shuffle.partitions本质上决定了Shuffle后数据的分散程度，具体数字要结合数据量和Executor数匹配。比如40个Executor、每个4核，总并发约160个slot，Shuffle分区200能保证每个slot都有活干；如果total并发大于分区数，说明分区数量反而变成性能瓶颈了。

5.2 Spark on YARN，队列隔离与重试机制

在大多数公司，Spark作业都运行在YARN集群上。双11这种大促期间，同一个YARN队列会有成百上千个作业在跑，如果不做资源隔离和优先级控制，核心作业可能被一些非重要任务挤掉。

我们的策略很简单：把核心分析作业放到高优先级队列，非核心作业放到共享队列，通过YARN的capacity调度器做资源隔离。核心队列保证至少占集群50%的算力，大促期间非核心任务即使排队，也不能影响核心产出。

作业重试机制也不能忽视。双11凌晨有上百个分析作业互相依赖，一个失败会导致下游一串失败。我们在调度系统里给每个作业配置了失败自动重试，依赖关系按小时等待。注意重试不是无脑叠加，最多重试3次，超过则发送告警，由值班人工介入。事实证明大促期间有些作业失败是瞬时的，比如NameNode抖动、YARN队列被占满，几分钟后自己重试就能成功，重试机制帮我们避免了很多夜间电话叫醒。

5.3 结果落库的隐性坑

分析结果最后要写到关系型数据库或者Hive的ADS层结果表。这里有个常见坑：小批量数据多次写入，在MySQL这类数据库会产生大量小事务，直接影响线上业务数据库的稳定性。

我们统一做法是：把Spark计算结果物化到HDFS上的结果表中，再由专门的同步任务批量写入到线上库。写库前做一次去重和幂等控制，以“业务日期+维度”作为唯一主键，重复跑数不会产生重复数据。这步看起来跟Spark没啥关系，但后期95%的线上数据质量问题都是因为这步没做好而造成的。

6. 跑数验证、结果链路与复盘心得

6.1 和线上看板对账，永远是第一优先级

离线分析和实时分析用的是两套链路，数据通常会有差异。差异来源包括：

• 实时计算窗口长度与离线统计窗口不一致；
• 实时采用近似去重，离线精确去重；
• 实时数据有流量采样。

我们在每次发布前，都会拿离线结果和实时大屏的数据做一次对账。正常情况下误差在2%以内是允许的，如果超过这个数，就要优先检查口径是不是统一。

这里有个实用的技巧：不要用“指标最终值”对比，而是直接用“底表明细条数”对比。对比总订单数、总支付金额，再下钻到每个小时段看差异曲线，哪一段出现尖刺，说明那段数据链路的某个环节可能出了问题，对照定位会非常高效。

6.2 从结果表到报表展示的数据链路

Spark分析完成后，最终结果会落到ADS层数据表里。报表系统和数据大屏只需要定时读取这些结果表，不需要感知背后复杂的数据加工逻辑。

结果表设计上要预留扩展性。比如双11期间运营可能需要实时看每个小时的数据，我们的结果表就会细化到小时颗粒度，同时保留当天累计值，避免报表层多表拼装。ADS表统一用这个结构：

• 日期、小时
• 维度值（品类、品牌、地区等）
• 各指标值
• 更新时间

这个设计让报表侧对接非常简单，无论是要单日聚合、小时趋势还是维度下钻，都能从一张表出数。

6.3 复盘：这类项目还有哪些可以延伸的方向

Spark分析篇到这里算是完整走了一遍，但双11大数据分析这个命题不只于此。实测下来有几个方向非常值得继续投入：

一是实时召回复核。离线结果第二天才能出，大促当天运营看的是实时数据。可以把实时计算结果每小时落一次快照，批量分析时对比实时链路与离线链路的差值，反哺修正实时计算逻辑，让实时数据越来越接近精确值。

二是细分人群分析。当前更多是基于全量和品类维度的分析，还可以进一步加工用户标签：新客老客、高活跃低活跃、价格敏感型、品牌偏好型等，让分析从“看大盘”走进“理解人”，业务价值高出一个量级。

三是数据回填和补数自动化。双11这种极端场景，凌晨ETL容易出现上游数据延迟、作业失败等情况。可以构建一套自动补数流程，根据上游表分区是否就绪来判断下游是否需要等待，而不是让作业直接失败。这块我们后期做成了一个数据质量校验的服务，对调度稳定性的帮助极大。

最后再分享一个我个人的实操习惯：大促分析项目跑之前，一定要准备一份“数据地图”，把所有依赖表的来源、负责人、产出时间、分区字段列清楚。双11当天大家压力都很大，凌晨两点数据没产出，如果没有地图，排查问题就像无头苍蝇。有了地图，能直接定位到是该找上游，还是该看自己的代码，问题解决效率完全不同。