共计 2071 个字符，预计需要花费 6 分钟才能阅读完成。
背景与痛点
电商平台每天产生海量的用户行为数据，包括浏览、收藏、加购、购买等。这些数据蕴含着用户偏好、购买意向等宝贵信息，但处理起来面临三大挑战：
- 数据规模大：原始数据集超过 100GB，单机处理效率低下
- 维度复杂：包含用户 ID、商品 ID、行为类型、时间戳等多个维度
- 时效性要求高：需要近实时处理才能发挥最大商业价值
技术选型对比
我们对比了三种常见工具在处理该数据集时的表现（测试环境：8 核 CPU/32GB 内存）：
| 工具 | 加载时间 | 内存峰值 | 适合场景 |
|---|---|---|---|
| Pandas | 42 分钟 | OOM | 小规模数据分析 |
| Dask | 18 分钟 | 28GB | 中等规模数据 |
| PySpark | 6 分钟 | 12GB | 分布式大数据处理 |
PySpark 凭借其分布式计算能力成为最佳选择。
核心实现
数据加载与预处理
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
# 初始化 Spark 会话
spark = SparkSession.builder \
.appName("taobao_analysis") \
.config("spark.executor.memory", "8g") \
.getOrCreate()
# 加载 CSV 数据
df = spark.read.csv("taobao_2017.csv",
header=True,
inferSchema=True)
# 数据清洗示例：过滤异常时间戳
df_clean = df.filter((col("time_stamp") >= "2017-01-01") &
(col("time_stamp") <= "2017-12-31"))RFM 特征工程
# Recency 计算：用户最近一次行为距今的天数
recency = df.groupBy("user_id") \
.agg(max("time_stamp").alias("last_activity"))
.withColumn("recency",
datediff(current_date(), col("last_activity")))
# Frequency 计算：用户总行为次数
frequency = df.groupBy("user_id").count().withColumnRenamed("count", "frequency")
# Monetary 计算：用户总消费金额
monetary = df.filter(col("behavior_type") == "buy") \
.groupBy("user_id") \
.agg(sum("price").alias("monetary"))FP-Growth 算法实现
from pyspark.ml.fpm import FPGrowth
# 将用户行为序列转换为 itemsets
user_actions = df.groupBy("user_id") \
.agg(collect_list("item_id").alias("items"))
# 训练 FP-Growth 模型
fpGrowth = FPGrowth(itemsCol="items",
minSupport=0.01,
minConfidence=0.5)
model = fpGrowth.fit(user_actions)
# 查看频繁项集
model.freqItemsets.show(5)性能优化
- 分区策略：
- 按用户 ID 哈希分区避免数据倾斜
-
设置合理分区数（建议为 CPU 核数的 2 - 3 倍）
-
内存管理：
- 对于小表使用 broadcast join
-
及时调用
unpersist()释放缓存 -
并行度调优：
spark.conf.set("spark.default.parallelism", "48") spark.conf.set("spark.sql.shuffle.partitions", "48")
生产环境避坑指南
数据倾斜解决方案
当发现某个 task 执行时间异常长时：
-
识别倾斜 key：
df.groupBy("user_id").count().orderBy("count", ascending=False).show(10) -
解决方案：
- 对倾斜 key 单独处理
- 使用 salting 技术（添加随机前缀）
时间窗口处理
计算 7 天滚动窗口时要注意：
- 使用
window函数而非groupBy - 设置合理的 watermark 防止内存溢出
windowed = df.withWatermark("time_stamp", "1 hour") \
.groupBy(window("time_stamp", "7 days"), "user_id") \
.count()总结与扩展
通过本次分析，我们实现了：
- 用户价值分层（RFM 模型）
- 商品关联规则挖掘（FP-Growth）
- 实时处理流水线搭建
这些结果可直接用于：
- 个性化推荐系统
- 精准营销活动
- 库存预警系统
下一步可以尝试：
- 使用 GraphFrames 构建用户关系图
- 应用深度学习模型（如 NCF）进行 CTR 预测
- 结合实时流处理（Spark Streaming）
完整代码已上传 GitHub，包含详细注释和测试数据。在实际业务中应用时，记得根据具体场景调整参数阈值。
正文完
