Kmeans算法来实现RFM指标计算步骤

        K-Means（K均值）是一种经典的无监督聚类算法，主要用于将数据集划分为 KKK 个不同的簇（Cluster）。

        它基于最小化簇内样本的平方误差，即最小化数据点与簇中心的距离之和。

---

1. K-Means 算法原理

(1) 主要步骤

1. 初始化
   选择 KKK 个初始聚类中心（可以随机选取，也可以使用K-Means++优化选择）。

2. 分配样本到最近的簇中心
   对于数据集中的每个点，将其分配给最近的聚类中心（使用欧几里得距离、曼哈顿距离等度量）。

3. 更新簇中心
   计算每个簇中所有点的均值，并将该均值作为新的簇中心。

4. 重复步骤 2 和 3
   直到聚类中心不再发生变化或者达到最大迭代次数。

(2) 目标函数

K-Means 试图最小化以下目标函数（平方误差）：

                     

---

2. K-Means 的特点

(1) 优势

• 计算速度快，时间复杂度接近 O(nKT)O(nKT)O(nKT)（其中 TTT 是迭代次数）。
• 适用于大规模数据集，易于并行化（可以结合MapReduce/Spark实现）。
• 聚类结果可解释性强，适用于数据分析、推荐系统、用户分群等场景。

(2) 局限性

• K值需要人为指定，如果选择不当，可能会导致聚类效果不佳。
• 对初始簇中心敏感，可能陷入局部最优（K-Means++ 可以优化初始中心选择）。
• 对异常点敏感，受离群点影响较大。
• 适用于凸形簇，对非球形分布的数据表现较差。

3. K-Means Demo示例

# 使用机器学习算法实现rfm
from utils.base import BasicTag
from pyspark.sql import SparkSession,functions as F
from pyspark.sql.types import *
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.clustering import KMeansclass RFMTag(BasicTag):def make_tag(self):rule_dict = self.data_dict# 1-计算获取RFM的值# 消费周期计算recency = F.datediff(F.current_date(), F.from_unixtime(F.max('finishtime'))).alias('diff_dt').alias('recency')# 订单量frequency = F.countDistinct('ordersn').alias('frequency')# 订单金额monetary = F.sum('orderamount').alias('monetary')df_rfm = self.df_es.groupby('memberid').agg(recency,frequency,monetary)# 2-根据计算的结果，将数据划分5个等级进行打分
# -----------------------------R数据打分----------------------------------------------------------------------------------# 2-1 特征工程将数据转为向量(只有转化为坐标向量（VectorAssembler），才能在坐标轴上计算方向及两点之间的距离，再求平均距离)v_r = VectorAssembler(inputCols=['recency'],outputCol='v_r')# 添加转化数据，将上面订单金额加入进行转化df_r_fit = v_r.transform(df_rfm)# 2-2 使用kmeas算法打分，#     k参数负责划分几个分数段；#     featuresCol负责传入向量值用于聚类的特征向量的列名；#     predictionCol 用于存储聚类结果的列名。KMeans算法会在训练数据上进行聚类分析，并将每个数据点分配到一个聚类。这个聚类结果会存储在predictionCol指定的列中。#+---+-----------------+                                    +---+-----------------------+#|id |        v_r      |                                    |id |        v_r      | k_r |#+---+-----------------+                                    +---+-----------------+-----+#| 1 | [1.0, 2.0, 3.0] |          执行KMeans聚类后：          | 1 | [1.0, 2.0, 3.0] |  0  |#| 2 | [4.0, 5.0, 6.0] |    ————————————————————————>       | 2 | [4.0, 5.0, 6.0] |  1  |#| 3 | [7.0, 8.0, 9.0] |                                    | 3 | [7.0, 8.0, 9.0] |  2  |#| 4 |[10.0, 11.0, 12.0]|                                   | 4 |[10.0, 11.0, 12.0]|  3 |kmeans = KMeans(k=5,featuresCol='v_r',predictionCol='k_r')# 添加数据----kmeans.fit( )用于训练模型，训练完生成一个KMeans模型对象k_r_fitk_r_fit = kmeans.fit(df_r_fit)# 转化数据,对训练结果进行预测，将预测结果添加到数据框：df_k_r中df_k_r = k_r_fit.transform(df_r_fit)df_k_r.show()# 计算k值中心判断k值的大小--clusterCenters()获取每个聚类的中心点（质心），存储在cluster变量中。cluster = k_r_fit.clusterCenters()# 遍历k值（质点）中心，计算k值的和# cluster_dict = {}：创建一个空字典，用于存储每个聚类中心点的索引和其特征值的和。# enumerate：是 Python 的一个内置函数，用于在遍历可迭代对象（如列表、元组、字符串等）时，同时获取每个元素的索引和值。它返回一个枚举对象，该对象生成一系列包含索引和值的元组。# 示例列表：#------------------------------------------+#   fruits = ['apple', 'banana', 'cherry'] |#   #使用 enumerate 进行遍历                 |#   for index, value in enumerate(fruits): |#   rint(index, value)                     |#--------输出------------------------------+#   0 apple                               |#   1 banana                              |#   2 cherry                              |#-----------------------------------------+cluster_dict = {}for i, v in enumerate(cluster):cluster_dict[i] = sum(v)print(cluster_dict)  # 输出结果:{0: 1445.0, 1: 1446.0, 2: 1447.0}# 中心点排序，对cluster_dict按特征值的和进行升序排序--并强制类型转化为字典格式cluster_dict_sort = dict(sorted(cluster_dict.items(), key=lambda x: x[1]))# 对k中心替换大小# count = 0：初始化计数器。# for k, v in cluster_dict_sort.items()：遍历排序后的字典cluster_dict_sort。# cluster_dict_sort[k] = count：将排序后的中心点索引替换为从0开始的递增值。# count += 1：计数器递增。然后使用cluster_dict_sort[k] 进行值的替换count = 0for k, v in cluster_dict_sort.items():cluster_dict_sort[k] = countcount += 1# 替换原来df中的k值编号@F.udf(returnType=IntegerType())def repace_data(k):return cluster_dict_sort.get(k)df_replace_r = df_k_r.select(df_k_r.memberid,df_k_r.recency,df_k_r.frequency,df_k_r.monetary,repace_data('k_r').alias('k_r'))#-----------------------------F数据打分----------------------------------------------------------------------------------# 2-1 特征工程将数据转为向量f_r = VectorAssembler(inputCols=['frequency'], outputCol='f_r')# 添加转化数据df_f_fit = f_r.transform(df_replace_r)# 2-2 kmeas算法打分kmeans = KMeans(k=5, featuresCol='f_r', predictionCol='k_f')# 添加数据f_r_fit = kmeans.fit(df_f_fit)# 转化数据df_f_r = f_r_fit.transform(df_f_fit)# 计算k值中心判断k值的大小cluster = f_r_fit.clusterCenters()# 遍历k值中心，计算k值的和cluster_dict = {}for i, v in enumerate(cluster):cluster_dict[i] = sum(v)# 中心点排序cluster_dict_sort = dict(sorted(cluster_dict.items(), key=lambda x: x[1]))# 对k中心替换大小count = 0for k, v in cluster_dict_sort.items():cluster_dict_sort[k] = countcount += 1# 替换原来df中的k值编号@F.udf(returnType=IntegerType())def repace_data1(k):return cluster_dict_sort.get(k)df_replace_f = df_f_r.select(df_f_r.memberid, df_f_r.recency, df_f_r.frequency, df_f_r.monetary,df_f_r.k_r,repace_data1('k_f').alias('k_f'))print('F计算完成')
# -----------------------------M数据打分----------------------------------------------------------------------------------# 2-1 特征工程将数据转为向量# TODO:inputColss输入字段修改  outputCol字段修改m_r = VectorAssembler(inputCols=['monetary'], outputCol='m_r')# 添加转化数据# TODO: 替换上一步的df数据df_f_fit = m_r.transform(df_replace_f)# 2-2 kmeas算法打分# TODO:inputColss输入字段修改  outputCol字段修改kmeans = KMeans(k=5, featuresCol='m_r', predictionCol='k_m')# 添加数据f_r_fit = kmeans.fit(df_f_fit)# 转化数据df_f_r = f_r_fit.transform(df_f_fit)df_f_r.show()print('M评分')# 计算k值中心判断k值的大小cluster = f_r_fit.clusterCenters()# 遍历k值中心，计算k值的和cluster_dict = {}for i, v in enumerate(cluster):cluster_dict[i] = sum(v)# 中心点排序cluster_dict_sort = dict(sorted(cluster_dict.items(), key=lambda x: x[1]))# 对k中心替换大小count = 0for k, v in cluster_dict_sort.items():cluster_dict_sort[k] = countcount += 1# 替换原来df中的k值编号@F.udf(returnType=IntegerType())def repace_data2(k):return cluster_dict_sort.get(k)# TODO:df_replace_m = df_f_r.select(df_f_r.memberid, df_f_r.recency, df_f_r.frequency, df_f_r.monetary, df_f_r.k_r, df_f_r.k_f,repace_data2('k_m').alias('k_m'))df_replace_m.show()print('M计算完成')# 3-计算所有用户的平均打分--结果为一行数据df_rfm_score_avg = df_replace_m.select(F.avg('k_r').alias('r_avg'),F.avg('k_f').alias('f_avg'),F.avg('k_m').alias('m_avg'))# +-----+-----+-----+# |r_avg|f_avg|m_avg|# +-----+-----+-----+# | 2.0 | 3.0 | 4.0 |# +-----+-----+-----+# 将平均分转为row对象可以获第一行取值(实际上只有一行)# row 是一个 Row 对象，它允许你像访问属性一样访问列值。row = df_rfm_score_avg.first()# 4-通过平均值判断rfm的高低df_rfm_num =  df_replace_m.select(df_replace_m.memberid,# 大于为1  否则为0F.when(df_replace_m.k_r >= row['r_avg'],1).otherwise(0).alias('rn'),F.when(df_replace_m.k_f > row['f_avg'],1).otherwise(0).alias('fn'),F.when(df_replace_m.k_m > row['m_avg'],1).otherwise(0).alias('mn'),)# 5-标签匹配# df_rfm_num 是一个包含 RFM 分数的 DataFrame。# select 方法选择了 memberid 列和通过 concat 方法连接的 RFM 分数（即 rn、fn、mn）。# F.concat 将 R、F、M 分数转化为字符串并连接成一个新的字符串。# 最后，alias('rfm') 为这个新字符串列命名为 rfmdf_rfm_str = (df_rfm_num.select(df_rfm_num.memberid,F.concat  # F.concat 将 R、F、M 分数转化为字符串并连接成一个新的字符串。并且alias('rfm') 为这个新字符串列命名为 rfm(df_rfm_num.rn.cast('string'),df_rfm_num.fn.cast('string'),df_rfm_num.mn.cast('string')).alias('rfm')))df_rfm_str.show()# 这是一个用户定义函数（UDF），用于根据 rule_dict 字典将 RFM 字符串映射到相应的标签。# @F.udf(returnType=StringType()) 装饰器用于将 Python 函数转化为 PySpark UDF，并指定返回类型为 StringType()。@F.udf(returnType=StringType())def match(data):return rule_dict.get(data)# 应用 match 函数并创建新 DataFrame：self.df_new_tag = (df_rfm_str.select(df_rfm_str.memberid.alias('userID'),match(df_rfm_str.rfm).alias('tagsID')))# 创建对象
ss = SparkSession.builder.config('spark.sql.shuffle.partitions','6').getOrCreate()
rfm = RFMTag()
# 执行
rfm.action('RFM', ss, 'tfec_tags', 'tbl_basic_tag')