基于Spark的零售商户生存分析与风险画像实践

我之前接过一个类似的零售数据分析项目，数据量到了日增几百万笔交易的地步，用传统单机Pandas跑特征聚合，一个全量汇总要跑四十多分钟，偶尔内存直接爆掉。换到Spark之后，同一个任务从四十多分钟压到三分钟以内，而且横向扩容非常轻松。这篇文章把整个项目从数据口径定义、Spark算子设计、生存分析模型、机器学习风险画像到可视化大屏落地的全过程拆开来讲，附带关键源码片段和部署过程中踩过的坑。

1. 分析口径先行：零售交易者数据到底怎么定义"活着"和"死亡"

很多第一次做这类项目的人容易犯一个错误：拿到数据就开始写代码，跑出一堆图表，结果业务方问一句"你说的活跃率是怎么定义的"，当场答不上来。所以这个项目我做的第一件事，不是搭Spark集群，而是跟业务方一起把口径敲死。

1.1 零售交易者：商户、门店还是消费者账户

"零售交易者"这个词在不同的业务体系里指向完全不同。在电商平台里，交易者可能是买家账户；在线下连锁零售体系里，交易者往往是加盟商、门店或者独立商户；在供应链金融场景里，交易者还可能是小微经销商。

我们这个项目面对的是连锁零售体系下的门店和独立商户，以每笔POS交易流水和线上订单为载体。分析目标是判断这些门店/商户在过去一段时间内的经营活跃度、交易稳定性以及未来流失/经营恶化的风险，也就是标题里说的"行为特征"和"生存状况"。

之所以强调这个前提，是因为后面的特征工程、生存分析口径、风险标签定义，全是围绕"门店/商户作为交易主体"来展开的。如果你换到消费者画像场景，很多指标要推倒重来。

1.2 核心口径三件套：出生、观察期、沉默判定

生存分析里最经典的三个问题：个体从什么时候开始"暴露在风险中"？什么时候算"死亡"？观察期截止到什么时候？放到零售商户场景下，我定义的口径如下：

• 出生时间：门店/商户产生第一笔有效交易订单的日期，记为first_trade_date。注意是"有效交易"，退款、作废、测试单都要先过滤掉。

• 死亡事件：超过设定阈值天数没有任何有效交易。阈值取多少很关键，我见过有人拍脑袋定30天，结果把季节性强、本来就有淡旺季的门店误判为"死亡"。这个项目里我对比了三种阈值（14天、30天、60天）下的结果差异，最后选了30天作为主口径，60天作为风险预警口径。

• 观察期：分析窗口的结束日期，一般是当前日期或者指定的数据截止日期。生存时间的计算要区分"已死亡"和"右删失/截尾"两种情况：已经沉默超过30天的门店，生存时间记为last_trade_date - first_trade_date；还没有沉默的门店，生存时间记为obs_date - first_trade_date，并且标记为删失。

这里有一个很关键的点：右删失（right-censored）样本不能当普通死亡样本处理。如果用二分类模型直接预测"死亡概率"，把所有未沉默样本当负样本，模型学到的其实是"历史活跃长度"而不是"未来失活风险"，落地效果会偏得厉害。生存分析里的KM曲线和Cox模型恰恰是为了处理这种删失数据而生的。

1.3 数据形态与主键设计

项目里最核心的两张原始表：

• 交易流水表：每笔订单一行，字段包括store_id、trade_date、order_id、amount、channel、sku_category、discount、refund_flag等。

• 门店档案表：每个门店一行，字段包括store_id、open_date、province、city、store_type（直营/加盟/合作）、area（商圈属性）等。

分析主键是store_id，所有行为特征、生存状态、风险评分都挂在门店粒度上。大促期间的虚假订单、刷单行为，交易特征会发生剧烈异常，这也是后面风险画像的一个重要信号来源。

2. 特征宽表构建：从Spark读取JSON交易流水到聚合特征的关键路径

数据底座的核心产物是一张"门店 × 时间窗口 × 特征"的宽表。所谓特征宽表，就是把原始明细数据按分析主键聚合，然后横向拼接成一行一行的特征向量，供后续统计分析和机器学习直接使用。

2.1 Spark读取JSON流水的三种方式

交易数据在我们这里不是CSV，而是日志采集系统导出的JSON文件，每天一批，按日期分区存放。Spark读JSON非常方便，关键是把嵌套结构和类型推断处理好。

第一种方式，直接用spark.read.json("path")，Spark会根据JSON字段自动推断Schema。好处是省事，坏处是如果某些字段偶尔缺失或者类型不稳定，推断结果会让人抓狂，比如金额字段有时候是数字有时候是字符串。

第二种方式，先定义好Schema再读，推荐在生产环境中用：

import org.apache.spark.sql.types._ val tradeSchema = StructType(Array( StructField("store_id", StringType, true), StructField("trade_date", StringType, true), StructField("order_id", StringType, true), StructField("amount", DoubleType, true), StructField("discount", DoubleType, true), StructField("channel", StringType, true), StructField("sku_category", StringType, true), StructField("refund_flag", IntegerType, true), StructField("event_time", StringType, true) )) val rawDF = spark.read.schema(tradeSchema).json("hdfs://namenode:8020/data/trade/2024/05/")

显式定义Schema之后，数据清洗阶段会省掉大量因为"字段类型推导不一致"造成的麻烦。再加上一个partitionBy分区字段，把trade_date拆成年月日目录，查询的时候可以直接做分区裁剪。

第三种方式，如果JSON是gz压缩后的，也不需要额外处理，Spark的json数据源会自动识别并解压。这里有个小经验：不要把数据文件搞得太小，比如几万个小文件放在一个目录下，Spark读的时候要生成大量task，调度成本比计算成本还高。后面专门有一节讲小文件问题。

2.2 清洗阶段：先过滤坏数据再做聚合

加载进来之后，第一轮清洗绝对不能省。我通常按这个顺序来：

1. 过滤store_id为空、amount为null或amount <= 0的记录；
2. 过滤退款标志位异常的数据（除非专门做退款分析）；
3. 去掉测试门店和白名单门店，一般业务方会维护一份test_store_list；
4. 识别并剔除明显刷单的极端交易记录，比如同一门店同一秒内上百笔相同金额订单，这种数据特征提取时容易被误判为"冲量能力强"。

清洗之后，再进入聚合环节。

2.3 从明细到宽表的聚合策略

特征宽表按时间窗口拆分为三个粒度：最近7天特征、最近30天特征、历史全量统计特征。这样做的好处是：短期特征捕捉最近波动，中期特征反映稳定趋势，历史特征描述整体等级，三个维度叠起来模型信息量更足。

每个窗口下常见的聚合逻辑：

在Spark代码里，最典型的宽表构建是先用groupBy做分层聚合，再通过多个when条件生成衍生指标：

import org.apache.spark.sql.functions._ import org.apache.spark.sql.expressions.Window val storeStats = rawDF .filter(col("trade_date") >= "2024-04-01") .groupBy("store_id") .agg( count("order_id").alias("order_cnt_30d"), sum("amount").alias("gmv_30d"), avg("amount").alias("avg_order_amount_30d"), countDistinct("trade_date").alias("active_days_30d"), countDistinct("sku_category").alias("category_cnt_30d"), avg(when(col("discount") > 0, col("discount")).otherwise(lit(0))).alias("avg_discount_30d"), sum(when(col("refund_flag") === 1, 1).otherwise(0)).alias("refund_cnt_30d") )

多窗口特征拼起来之后，store_id为主键，慢慢往宽表里join其他特征表。只要主键不重复，拼接顺序不影响结果。

2.4 窗口特征的泄漏问题

这是机器学习建模时最容易出问题的环节。所谓标签泄漏，是指特征里包含了只有"未来"才能知道的信息，导致模型在训练集上的表现虚高，上线之后立刻崩盘。

举个例子：我们要预测一个门店未来30天是否沉默，于是把"过去30天是否已经沉默"作为特征放进去。训练的时候模型发现这个特征和标签高度相关，AUC高达0.98，但上线时新样本还没有未来标签，这个特征的值要么为Null要么全为0，模型直接失效。

正确做法是：特征窗口和标签窗口在时间上必须严格错开。比如用4月1日之前的30天数据做特征，用5月1日的状态做标签，中间留出一天以上的间隔。项目里我统一约定：特征截止日为T，标签观察日为T+30，特征里严禁出现任何与T+1之后数据相关的字段。

3. 生存曲线与行为特征：在Spark中算出留存、活跃周期和沉默风险

前面说过，这个项目的核心分析目标之一是"生存状况"。这里要用到生存分析的思想和方法，用它来回答两个问题：门店从开业到流失，中位生存期是多长？什么样的行为特征组合会显著拉长或缩短生存期？

3.1 为什么选生存分析，而不是普通留存率

普通留存率只统计"第N天还有多少比例的人在活跃"，它有两个缺陷：一是没有考虑个体加入时间不同导致的队列偏差；二是把所有流失样本和无流失样本混在一起，信息利用不充分。

生存分析里的Kaplan-Meier方法则利用所有样本的存活时长和删失状态，能估计出"随时间变化的存活概率曲线"。在零售门店场景里，这个曲线直接告诉业务方：假设有100家新店同时开业，第180天大概还有多少家保持活跃。

在Spark里构造生存分析需要的三列数据：duration（存活天数）、event（是否发生了死亡事件，1为死亡，0为删失）、group（分组字段，比如门店类型）。

3.2 用Spark构造生存分析样本

第一步，算出每个门店的首末交易日期和死亡标记：

val lifeDF = spark.sql(""" SELECT store_id, first_trade_date, last_trade_date, obs_date, if(datediff(last_trade_date, first_trade_date) > silence_threshold, 0, 1) AS event, CASE WHEN datediff(last_trade_date, first_trade_date) > silence_threshold THEN datediff(obs_date, first_trade_date) ELSE datediff(last_trade_date, first_trade_date) END AS duration FROM store_trade_summary """)

这里silence_threshold按前面定的口径取30天。注意：event=1代表该门店在观察期内已经沉默（死亡），event=0代表到观察截止日仍然活跃（右删失）。

第二步，按存活天数分组，计算每个时间点上的存活概率。虽然KM曲线公式不复杂，但数据量大了以后分组聚合的计算量不小，Spark处理完全没压力：

val kmDF = lifeDF .groupBy("duration") .agg( count("*").alias("at_risk_start"), sum("event").alias("death_cnt") ) .orderBy("duration")

拿到每个时长上的风险集人数和死亡人数之后，用累积乘积公式算出存活率。这个计算可以拉回驱动端做，因为数据量通常只有几百行——你已经压缩到了"每个时长一行"。

3.3 按门店类型拆分的生存对比

分组KM曲线是业务方最想看的一张图：直营店和加盟店的存活曲线是否有显著差异。我按store_type分组分别计算了KM曲线，结论很直观：直营店由于有强运营支持，180天存活率在85%上下；加盟店波动更大，低线城市的加盟店180天存活率掉到60%以下。

这个分析的价值在于：它把"哪些门店风险高"从定性描述变成了可量化的生存概率。后续机器学习风险画像的特征里，门店类型、城市等级、开业时长本身就是重要维度，逻辑上正好闭环了。

3.4 行为特征对生存时间的影响：先把底层事实算出来

到这里，Spark主要承担的是一个"大规模事实计算器"的角色。我们不需要直接在Spark里跑复杂的Cox回归（除非数据量大到单机放不下，这个项目里暂时还没到这个量级），而是先通过Spark把每个门店的行为特征宽表和生存状态样本准备好，再交给Python端做统计建模。

算出来之后有几个值得业务关注的行为事实：

• 开业后前30天日均单量低于5单的门店，6个月内的沉默率是其他门店的2.3倍；
• 退款率超过15%的门店，存活率呈现明显下探；
• 品类结构过于单一（核心品类占比超过80%）的门店，季节性波动大，生存风险随之增高；
• 夜间交易占比（晚10点后交易占比）异常偏高的门店，要警惕非正常经营行为。

这些发现不一定全部进入最终模型，但作为业务洞察输出到可视化看板上，比单纯一堆模型系数更能获得业务方信任。

4. 机器学习风险画像：标签定义、模型训练与评分落地

光有生存曲线还不够，业务方需要知道"现在还在营业的门店里，哪些已经处于高风险状态"。这块就是风险画像：通过机器学习模型给每个门店输出一个风险评分，再按分数段划分风险等级。

4.1 标签怎么定

风险模型的标签定义是整个建模流程里最容易扯皮的地方。我建议不要自己拍脑袋，而是先把统计口径列出来跟业务方对齐。项目里采用的口径如下：

• 正样本（高风险）：观察窗口（比如2024年6月）内满足"连续30天以上无有效交易"且"之后60天内没有恢复迹象"的门店，标记为1。
• 负样本（低风险）：观察窗口内交易活跃、且未来90天仍有持续交易的门店，标记为0。
• 灰色样本：暂时沉默但未满30天、或中间有过零星交易的门店，建模时先剔除或者单独分桶。

这一步很关键：不要把灰色样本硬塞进0或1，否则模型学出来的边界会非常模糊。灰色样本可以后续用来做"预警名单"。

4.2 特征工程进阶：RFM变体与行为稳定性

除了宽表里已经算好的基础特征，风险画像模型还加入了三个方向的特征：

• 变体时长特征：最近一笔交易距今多少天、最近连续活跃天数、最长连续活跃天数。这些特征是"沉默风险"的最直接信号。
• 稳定性特征：日交易额变异系数（CV）、月销售额直线回归斜率、峰值日交易额与中位日交易额之比。CV越大说明交易越脉冲式，经营越不稳定。
• 异常模式特征：退款率、平均折扣率、极小金额订单占比、同一订单号重复使用次数。

特征做出来之后一定要做相关性筛查。项目里初始候选特征有80多个，做了两轮筛选之后保留30个左右。筛选逻辑不只看单特征AUC，还要看特征之间的相关性和业务可解释性。

4.3 模型方案：先是Spark MLlib，后来切到XGBoost

这里说下技术选型的演进。项目第一阶段用的是Spark MLlib自带的LogisticRegression，因为它跟整个Spark流水线无缝衔接，特征宽表算完直接训练，不需要搬数据，部署也简单。LR模型在AUC差不多0.78左右，胜在稳定可解释，特征系数能直接转成业务规则。

第二阶段为了拉高区分度，我把特征宽表导出到Python侧，跑了XGBoost。换模型之后AUC提升到0.86左右，效果提升的主要来源是：XGBoost可以建模特征之间的非线性交互，比如"低GMV × 高退款率"这种组合风险模式，在LR里需要手动构造交叉特征才可能学到。

但这里有个平衡：XGBoost在Python侧训练，特征数据量不大的时候，直接从Spark导出到本地没问题；如果后续数据量上来了更规范的做法是通过Spark的applyInPandas或者spark-sklearn做分布式推断。我后面在"部署避坑"里会再提这个坑。

4.4 概率校准与风险分桶

二分类概率输出不能直接拿给业务方，他们不关心0.72还是0.68，他们关心的是"这家店要列入哪一级监控名单"。所以模型输出之后还要做校准和分桶：

• 先对预测概率做温度缩放或者等距分桶，观察分数分布是否单调——高风险段的预测概率应该明显高于低风险段。
• 然后按业务容忍度设三个阈值，比如p<0.3为低风险，0.3≤p<0.65为中等风险，p≥0.65为高风险。阈值的选择依据是混淆矩阵里的召回率和精确率。
• 落地时风险等级直接作为门店标签写回宽表，供可视化大屏读取展示。

Spark侧打分入库代码也很简单：

val scoredDF = featureDF.join(modelProbaDF, Seq("store_id"), "left") scoredDF .withColumn("risk_level", when(col("risk_prob") >= 0.65, "high") .when(col("risk_prob") >= 0.30, "medium") .otherwise("low")) .write.mode("overwrite").saveAsTable("risk_profile_result")

注意：这里的阈值只针对本项目数据分布，换一批门店区域或时间窗口后要重新校准。直接沿用别人的阈值是最容易翻车的做法。

5. 可视化大屏：指标口径、图表联动与刷新机制

模型算完之后，所有结果最终要落到可视化界面上，让业务方每天打开就能看到。可视化不是简单地画图，而是要回答三个问题：看什么指标？下钻到什么粒度？数据多久刷新一次？

5.1 技术选型：ECharts适配大数据量的正确姿势

可视化这块我选的是ECharts + 后端接口 + Redis缓存的方案。大屏前端就是一套HTML页面，用ECharts的散点图、折线图、地图、热力图组合展示。为什么不用Spark自带的GraphX或者Zeppelin？因为那是给分析人员做探索用的，不是给业务方每天盯盘用的。大屏需要的是稳定、好看、查询快。

但ECharts本身不适合直接渲染几十万点。所以落地时需要做数据下采样和聚合：地图散点展示门店粒度，最多同时展示3000个点；超过3000就要先从省级粒度下钻，再到城市，再到门店。下钻逻辑在后端SQL里完成，这样前端永远只拿少量聚合结果。

5.2 大屏指标设计

大屏顶部放四个核心KPI：总有效交易门店数、高风险门店数、中风险门店数、当天GMV。下方主体分三个模块：

• 左侧：风险门店分布Top省份地图 + 风险等级占比饼图。
• 中间：每日交易活跃门店趋势线 + 30天沉默率曲线。
• 右侧：高风险门店明细表格（Top20）+ 风险特征雷达图。

这里要注意指标对齐：地图上的门店数和明细表格里的门店数必须来自同一个结果表，否则业务方会发现"地图显示50家高风险，表格只有49家"，信任感直接崩塌。

5.3 表拖慢大屏？接口层做聚合和缓存

虽然有后端接口，但大屏页面上如果直接查询几十万行明细，数据库也扛不住。我采用的是三层方案：

1. 预计算结果表，每次离线跑批之后把日汇总结果写入聚合表，前端只查聚合表。
2. Redis缓存当日查询结果，TTL设成10分钟，大屏刷新时只打缓存，缓存过期后再回源查询一次。
3. 明细下钻接口限制返回条数，最多500条，超过就提示用户继续筛选。

如果要做实时刷新，就用WebSocket推送，后端每隔5分钟推一次最新数据。注意：实时刷新的更新频率和底层Spark离线计算的频次要协同，如果底层只每天算一次，前端再实时刷新也没有意义，反而让用户以为数据是实时的，产生误导。

5.4 可视化过程中的两个细节

一个是"存活率曲线"要用阶梯状图而不是平滑曲线。KM估计本身就是阶梯函数，平滑拟合反而丢失了真实信息。

另一个是风险等级颜色要统一：低风险绿色、中风险橙色、高风险红色，从表格、地图到饼图保持一致。这个看似简单，实际上前后端联调时特别容易因为色值不一致被业务方反复吐槽。

6. 集群部署与源码复现：内存调优、数据倾斜和小文件三大坑

这个项目做完之后，我总结出三个最折磨人的坑。如果你准备照着这个项目路线复现，先看完这一节，能省下几周的时间。

6.1 内存调优：shuffle时Executors频繁Full GC

项目初期跑30天聚合的时候，Spark任务时不时报OOM，日志里看到的主要现象是Shuffle spill和Full GC频发。后来定位下来原因有三层：

• Executor内存配比不对。默认spark.memory.fraction=0.6，但我们的作业以groupBy和join为主，shuffle数据占大头。我把spark.memory.fraction调到0.75，同时把spark.memory.storageFraction从默认0.5降到0.3，给执行内存留足了空间。
• spark.sql.shuffle.partitions默认200，对大规模数据不够。我们把它调到executor数 × 4，这个参数直接影响shuffle的并发度。
• 大表join小表时，把小表用broadcast join广播出去，避免全网shuffle。

最终配置大致如下：

spark-submit \ --master yarn \ --executor-memory 8g \ --num-executors 20 \ --executor-cores 4 \ --conf spark.memory.fraction=0.75 \ --conf spark.memory.storageFraction=0.3 \ --conf spark.sql.shuffle.partitions=80 \ --conf spark.sql.autoBroadcastJoinThreshold=10485760 \ --class com.retail.risk.SparkTradeFeatureJob \ retail-risk-1.0.0.jar

6.2 数据倾斜：集中在头部大店

门店交易数据天生有长尾特征：Top 5%的头部门店贡献了可能超过一半的交易量。按store_id做groupBy时，这几个头部门店对应的分区任务要处理的数据量是其他门店的几十倍，表现就是：某个task跑很久，其他task早已跑完在等它。

解决办法不外乎两种：

• 加盐（salting）：把热点key打散。比如给store_id加随机前缀或后缀，分成多个中间key分别聚合，最后再合并。
• 差异权重：如果数据倾斜来自join，尝试skew join优化，把大key拆开之后分别join。

在特征聚合阶段，我用的是第一种方案：先按store_id_hash % 10做一层分桶聚合，再按真实store_id汇总一次。整体优化后，作业时间从40分钟压到8分钟，效果非常明显。

6.3 小文件问题：结果表几万个小文件

写结果表的时候如果没控制分区粒度，会发现HDFS上多出来几万个小文件，每个几十KB，Spark读起来调度开销极大，Hive元数据也膨胀。这个问题在迭代过程中特别容易埋下雷。

我的做法是：写结果表之前先用repartition把分区数控制在一个合理范围，再按时间分区写入。

resultDF .repartition(col("dt")) .write .mode("overwrite") .partitionBy("dt") .format("parquet") .saveAsTable("store_feature_wide_table")

另外，spark.sql.adaptive.coalescePartitions.enabled开启后，Spark会在shuffle结束后自动合并小分区，这个参数在Spark 3.2+默认开启，但确认一下没坏处。

6.4 源码结构

项目源码的门类大致是这样的：

retail-risk/ ├── src/main/scala/com/retail/risk/ │ ├── SparkTradeFeatureJob.scala // 特征宽表构建主入口 │ ├── SurvivalSampleJob.scala // 生存分析样本生成 │ ├── RiskScoreJob.scala // 模型评分与风险分桶 │ └── utils/ │ ├── DateUtils.scala │ ├── SchemaProvider.scala // JSON Schema定义 │ └── DataCleaner.scala ├── scripts/ │ ├── submit_feature_job.sh │ ├── submit_survival_job.sh │ └── submit_score_job.sh ├── python/ │ ├── train_xgboost.py // 风险画像模型训练 │ ├── feature_importance.py │ └── threshold_calibration.py ├── dashboard/ │ ├── index.html // 大屏页面 │ └── dashboard.sql // 聚合结果查询SQL └── README.md

整个流程是：Spark作业产出特征宽表和生存样本 -> Python训练模型 -> 模型导出为pmml或者转成Spark可读格式 -> Spark评分作业打风险等级 -> 聚合结果写表 -> 大屏接口读取。

要复现的话，硬件环境建议至少3台4核16G的节点，纯单机跑Spark会非常吃力，其实现思路可以参考，但性能指标完全不可比。

7. 踩坑记录之外的一点个人体会

如果让我提炼这个项目里最值得反复咀嚼的经验，我觉得不是某个参数调优技巧，而是口径先行和数据质量这两个词。Spark再快，模型再高级，如果分析对象的口径定义不清、特征窗口和标签窗口混淆、历史数据里的脏数据没有被识别出来，输出的可视化看板再漂亮，也只是一堆无法落地的数字。

另一个体会是：大屏可视化不能做成模型的终点，而应该是数据产品闭环里的一环。模型产出风险等级后，应该打通到业务侧的预警工单、运营跟进流程，风险门店才能被真正干预。如果只看不做，分析就只是报表。

如果后续有余力，可以在现有框架上继续扩展几个方向：把特征宽表接入实时流计算（如Flink），做T+0甚至分钟级的风险预警；模型侧尝试生存森林（Random Survival Forest）替代固定窗口的二分类，让预测更贴近生存分析的本质；可视化侧考虑引入更多地理围栏和商圈竞争数据，提高风险识别的上下文丰富度。这些方向都可以复用本文的工程框架，核心的口径、进数、建模和展示链路只需要在局部做替换。