当前位置：首页 > news >正文

RFM 模型工程化落地：用户分层从 Excel 到自动化标签系统

news 2026/7/22 9:58:28

RFM 模型工程化落地：用户分层从 Excel 到自动化标签系统

大家好，我是朱大喜！今天聊一个数据分析师绕不开的经典话题——RFM 模型。别看它简单，真正要在生产环境中做到自动化、可维护、可回溯，里面的坑一个都不少。

一、从 Excel 手搓到自动化需求的诞生

RFM 模型大概是每个数据分析师入行时都会接触的东西。所谓 RFM，就是 Recency（最近一次消费时间）、Frequency（消费频率）、Monetary（消费金额）三个维度。用这三个维度给用户打分，然后做交叉分层，就能把用户分成"重要价值客户"、"重要挽回客户"、"一般维持客户"等不同等级。

刚开始做用户分层的时候，大家可能都经历过这个流程：从数据库导出一张 CSV，扔进 Excel，用 PERCENTILE 函数算分位点，VLOOKUP 分档，再手动标颜色。一周一次勉强能撑住，但要做到每天更新、多维度交叉、自动下发标签，Excel 就完全不够用了。

我们团队接到的需求是：为全平台 2000 万月活用户每天生成 RFM 标签，喂给下游的推荐系统和 CRM 平台。标签要求分层 5 档（R1-R5, F1-F5, M1-M5），每档的切割点基于前 90 天数据的百分位动态计算，还要支持按业务线（电商、内容付费、会员订阅）分别计算。

这个数据量和复杂度，决定了必须走系统化、工程化的路线。

二、RFM 指标计算的 SQL 实现

在数据仓库中，RFM 计算属于典型的窗口分析场景。我们用 Hive SQL 来实现每日增量计算。核心思路是：基于过去 90 天的订单数据，计算每个用户的 R、F、M 原始值，然后映射到 1-5 的标签档位。

-- ========== 第一步：计算用户级别的 R、F、M 原始值 ========== -- 基于过去90天的订单数据，按用户聚合计算 WITH user_rfm_raw AS ( SELECT user_id, -- R值：最近一次消费距今天数（越小越好，代表越活跃） DATEDIFF(CURRENT_DATE(), MAX(order_date)) AS recency_days, -- F值：近90天消费次数（越大越好） COUNT(DISTINCT order_id) AS frequency, -- M值：近90天消费总金额（越大越好） SUM(pay_amount) AS monetary, -- 业务线标识，用于分区计算分位点 business_line FROM dwd_order_detail WHERE order_date >= DATE_SUB(CURRENT_DATE(), 90) -- 取近90天数据 AND order_status = 'PAID' -- 只算已支付订单 AND pay_amount > 0 -- 过滤退款等异常数据 GROUP BY user_id, business_line ), -- ========== 第二步：按业务线计算分位点 ========== -- 使用 PERCENTILE 函数计算各指标在 20/40/60/80 分位的值 rfm_quantiles AS ( SELECT business_line, -- R 值分位点（注意 R 值反向：越小越好，所以高分段对应小值） PERCENTILE(recency_days, 0.2) AS r_p20, PERCENTILE(recency_days, 0.4) AS r_p40, PERCENTILE(recency_days, 0.6) AS r_p60, PERCENTILE(recency_days, 0.8) AS r_p80, -- F 值分位点 PERCENTILE(frequency, 0.2) AS f_p20, PERCENTILE(frequency, 0.4) AS f_p40, PERCENTILE(frequency, 0.6) AS f_p60, PERCENTILE(frequency, 0.8) AS f_p80, -- M 值分位点 PERCENTILE(monetary, 0.2) AS m_p20, PERCENTILE(monetary, 0.4) AS m_p40, PERCENTILE(monetary, 0.6) AS m_p60, PERCENTILE(monetary, 0.8) AS m_p80 FROM user_rfm_raw GROUP BY business_line ), -- ========== 第三步：映射 RFM 标签（1-5分） ========== user_rfm_label AS ( SELECT a.user_id, a.business_line, a.recency_days, a.frequency, a.monetary, -- R标签：recency越小越好，所以值越小分越高 CASE WHEN a.recency_days <= q.r_p20 THEN 5 WHEN a.recency_days <= q.r_p40 THEN 4 WHEN a.recency_days <= q.r_p60 THEN 3 WHEN a.recency_days <= q.r_p80 THEN 2 ELSE 1 END AS r_label, -- F标签：frequency越大越好 CASE WHEN a.frequency >= q.f_p80 THEN 5 WHEN a.frequency >= q.f_p60 THEN 4 WHEN a.frequency >= q.f_p40 THEN 3 WHEN a.frequency >= q.f_p20 THEN 2 ELSE 1 END AS f_label, -- M标签：monetary越大越好 CASE WHEN a.monetary >= q.m_p80 THEN 5 WHEN a.monetary >= q.m_p60 THEN 4 WHEN a.monetary >= q.m_p40 THEN 3 WHEN a.monetary >= q.m_p20 THEN 2 ELSE 1 END AS m_label, -- 组合标签：如 "R5_F4_M3" CONCAT('R', CASE WHEN a.recency_days <= q.r_p20 THEN '5' WHEN a.recency_days <= q.r_p40 THEN '4' WHEN a.recency_days <= q.r_p60 THEN '3' WHEN a.recency_days <= q.r_p80 THEN '2' ELSE '1' END, '_F', CASE WHEN a.frequency >= q.f_p80 THEN '5' WHEN a.frequency >= q.f_p60 THEN '4' WHEN a.frequency >= q.f_p40 THEN '3' WHEN a.frequency >= q.f_p20 THEN '2' ELSE '1' END, '_M', CASE WHEN a.monetary >= q.m_p80 THEN '5' WHEN a.monetary >= q.m_p60 THEN '4' WHEN a.monetary >= q.m_p40 THEN '3' WHEN a.monetary >= q.m_p20 THEN '2' ELSE '1' END ) AS rfm_label FROM user_rfm_raw a JOIN rfm_quantiles q ON a.business_line = q.business_line ) -- ========== 第四步：映射用户分层 ========== -- 将 RFM 标签组合映射为业务可理解的分层名称 SELECT user_id, business_line, rfm_label, CASE WHEN r_label >= 4 AND f_label >= 4 AND m_label >= 4 THEN '重要价值客户' WHEN r_label >= 4 AND f_label <= 2 AND m_label >= 4 THEN '重要挽回客户' WHEN r_label <= 2 AND f_label >= 4 AND m_label >= 4 THEN '重要保持客户' WHEN r_label >= 4 AND f_label >= 4 AND m_label <= 2 THEN '重要发展客户' WHEN r_label <= 2 AND f_label <= 2 AND m_label <= 2 THEN '流失客户' WHEN r_label >= 3 AND f_label >= 3 AND m_label >= 3 THEN '一般价值客户' ELSE '一般维持客户' END AS user_tier, CURRENT_TIMESTAMP() AS tag_generate_time FROM user_rfm_label;

这个 SQL 脚本每天跑一次，处理 2000 万用户的数据大约需要 8 分钟（Hive on Spark，集群 50 个 Executor）。分位点虽然用 PERCENTILE 函数有点暴力，但在百万级以上数据集上，误差通常在 0.5% 以内，业务完全可接受。

三、Python 调度脚本与自动化

SQL 算出了标签，但还需要调度编排、数据校验、异常告警一套流程。我们用 Python + Airflow 搭了一套完整的标签生产 Pipeline。

import pymysql from datetime import datetime, timedelta import smtplib from email.mime.text import MIMEText # ========== RFM 标签生产 Pipeline ========== class RFMTagPipeline: """RFM 标签自动化生产与校验流程""" def __init__(self, db_config): self.db_config = db_config self.conn = None def connect_db(self): """建立数据库连接""" self.conn = pymysql.connect(**self.db_config) def check_data_volume(self, target_date): """ 数据量校验：对比当日产出与近7日均值 如果偏差超过20%，触发告警 """ sql = """ SELECT COUNT(DISTINCT user_id) AS today_cnt, -- 计算近7天（不含今天）的日均用户数 (SELECT AVG(cnt) FROM ( SELECT COUNT(DISTINCT user_id) AS cnt FROM dwd_user_rfm_label WHERE dt >= DATE_SUB(%s, 7) AND dt < %s GROUP BY dt ) t) AS avg_7d_cnt FROM dwd_user_rfm_label WHERE dt = %s """ with self.conn.cursor() as cursor: cursor.execute(sql, (target_date, target_date, target_date)) today_cnt, avg_cnt = cursor.fetchone() # 偏差率计算 deviation = abs(today_cnt - avg_cnt) / avg_cnt if avg_cnt else 0 return today_cnt, deviation def check_distribution(self, target_date): """ 分层分布校验：各分层用户占比波动不超过5% 防止因上游数据问题导致标签大面积错乱 """ sql = """ SELECT user_tier, COUNT(*) AS user_cnt, COUNT(*) * 100.0 / SUM(COUNT(*)) OVER() AS pct FROM dwd_user_rfm_label WHERE dt = %s GROUP BY user_tier ORDER BY user_cnt DESC """ with self.conn.cursor() as cursor: cursor.execute(sql, (target_date,)) return cursor.fetchall() def send_alert(self, subject, content): """ 告警通知：企业微信机器人 / 邮件 """ # 使用企业微信 Webhook 发送告警 # 这里简化展示为邮件方式 msg = MIMEText(content, 'plain', 'utf-8') msg['Subject'] = f'[RFM标签告警] {subject}' msg['From'] = 'data-platform@company.com' msg['To'] = 'data-team@company.com' # smtp.send_message(msg) # 实际使用需配置 SMTP def run(self, target_date): """主流程：执行标签生产与校验""" print(f"[{datetime.now()}] 开始生产 {target_date} 的 RFM 标签...") self.connect_db() # 1. 执行 RFM SQL（由 Airflow 上游任务完成） # 2. 数据量校验 today_cnt, deviation = self.check_data_volume(target_date) print(f"今日标签用户数: {today_cnt}, 偏差率: {deviation:.2%}") if deviation > 0.2: # 超过20%偏差 self.send_alert( '数据量异常', f'{target_date} 标签用户数偏差 {deviation:.1%}，请检查上游数据！' ) raise ValueError(f"数据量偏差过大: {deviation:.2%}") # 3. 分布校验 distribution = self.check_distribution(target_date) print("各分层用户分布：") for tier, cnt, pct in distribution: print(f" {tier}: {cnt} ({pct:.1f}%)") # 4. 写入标签生产日志 log_sql = """ INSERT INTO rfm_tag_production_log (dt, total_users, status, create_time) VALUES (%s, %s, 'SUCCESS', NOW()) """ with self.conn.cursor() as cursor: cursor.execute(log_sql, (target_date, today_cnt)) self.conn.commit() print(f"[{datetime.now()}] {target_date} RFM 标签生产完成！") self.conn.close() # ========== 使用示例 ========== if __name__ == '__main__': db_config = { 'host': 'your-mysql-host', 'port': 3306, 'user': 'data_platform', 'password': '***', 'database': 'user_profile' } pipeline = RFMTagPipeline(db_config) # 通常由 Airflow 传入执行日期 target_date = '2026-07-22' pipeline.run(target_date)