如何从所有预测的集群中获取 n 个样本/事件并在 PySpark 中以数据框的形式返回？答案

【问题标题】：How can take n samples/events from all predicted clusters and return in form of dataframe in PySpark?如何从所有预测的集群中获取 n 个样本/事件并在 PySpark 中以数据框的形式返回？
【发布时间】：2021-10-29 12:10:30
【问题描述】：

我正在关注这个tutorial 并尝试挑选/选择顶部n 事件，假设在将预测的集群分配给主要df 之后有10 个事件/行并将它们合并并以以下形式报告火花数据框。假设我们有主数据框 df1 包含以下 3 个功能：

+-----+------------+----------+----------+
|   id|           x|         y|         z|
+-----+------------+----------+----------+
| row0|  -6.0776997|-2.9096103|-1.5181729|
| row1|  -1.0122601|  7.322841|-5.4424076|
| row2|   -8.297007| 6.3228936| 1.1672047|
| row3|  -3.5071216|  4.784812|-5.4449472|
| row4|   -5.122823|-3.3220499|-0.5069805|
| row5|  -2.4764006|  8.255791|  4.409478|
| row6|   7.3153954| -5.079449| -7.291215|
| row7|  -2.0167463|  9.303454|  7.095179|
| row8|  -0.2338185| -4.892681| 2.1228876|
| row9|    6.565442| -6.855994|-6.7983212|
|row10|  -5.6902847|-6.4827404|-0.9246967|
|row11|-0.017986143| 2.7632365| -8.814824|
|row12|  -6.9042625|-6.1491723|-3.5354295|
|row13|  -10.389865|  9.537853|  0.674591|
|row14|   3.9688683|-6.0467844| -5.462389|
|row15|   -7.337052|-3.7689247| -5.261122|
|row16|   -8.991589|  8.738728|  3.864116|
|row17| -0.18098584|  5.482743| -4.900118|
|row18|   3.3193955|-6.3573766| -6.978025|
|row19|  -2.0266335|-3.4171724|0.48218703|
+-----+------------+----------+----------+

现在我从聚类算法中获得了 datafarame df2 形式的信息，如下所示：

print("==========================Short report==================================== ")

n_clusters = model.summary.k
#n_clusters
print("Number of predicted clusters: " + str(n_clusters))

cluster_Sizes = model.summary.clusterSizes
#cluster_Sizes 

col = ['size']
df2 = pd.DataFrame(cluster_Sizes, columns=col).sort_values(by=['size'], ascending=True)  #sorting
cluster_Sizes = df2["size"].unique()
print("Size of predicted clusters: " + str(cluster_Sizes))
clusterSizes

#==========================Short report==================================== 
#Number of predicted clusters: 10
#Size of predicted clusters: [ 486  496  504  529  985  998  999 1003 2000]

+-----+----------+
|     |prediction|
+-----+----------+
|    2|       486|
|    6|       496|
|    0|       504|
|    8|       529|
|    5|       985|
|    9|       998|
|    7|       999|
|    3|      1003|
|    1|      2000|
|    4|      2000|
+-----+----------+

所以在这里，索引列是预测的集群标签。我可以将预测的集群标签分配到主数据框中，但不能分配集群大小，如下所示：

+-----+----------+------------+----------+----------+
|   id|prediction|           x|         y|         z|
+-----+----------+------------+----------+----------+
| row0|         9|  -6.0776997|-2.9096103|-1.5181729|
| row1|         4|  -1.0122601|  7.322841|-5.4424076|
| row2|         1|   -8.297007| 6.3228936| 1.1672047|
| row3|         4|  -3.5071216|  4.784812|-5.4449472|
| row4|         3|   -5.122823|-3.3220499|-0.5069805|
| row5|         1|  -2.4764006|  8.255791|  4.409478|
| row6|         5|   7.3153954| -5.079449| -7.291215|
| row7|         1|  -2.0167463|  9.303454|  7.095179|
| row8|         7|  -0.2338185| -4.892681| 2.1228876|
| row9|         5|    6.565442| -6.855994|-6.7983212|
|row10|         3|  -5.6902847|-6.4827404|-0.9246967|
|row11|         4|-0.017986143| 2.7632365| -8.814824|
|row12|         9|  -6.9042625|-6.1491723|-3.5354295|
|row13|         1|  -10.389865|  9.537853|  0.674591|
|row14|         2|   3.9688683|-6.0467844| -5.462389|
|row15|         9|   -7.337052|-3.7689247| -5.261122|
|row16|         1|   -8.991589|  8.738728|  3.864116|
|row17|         4| -0.18098584|  5.482743| -4.900118|
|row18|         2|   3.3193955|-6.3573766| -6.978025|
|row19|         7|  -2.0266335|-3.4171724|0.48218703|
+-----+----------+------------+----------+----------+

现在，希望通过以下函数以数据框的形式包含\报告每个集群的前 n 行。到目前为止我尝试过的是使用（多）条件过滤：

print("==========================Short report==================================== ")

n_clusters = model.summary.k
#n_clusters
print("Number of predicted clusters: " + str(n_clusters))

cluster_Sizes = model.summary.clusterSizes
#cluster_Sizes 

col = ['size']
clusterSizes = pd.DataFrame(cluster_Sizes, columns=col).sort_values(by=['size'], ascending=True)  #sorting
cluster_Sizes = clusterSizes["size"].unique()
print("Size of predicted clusters: " + str(cluster_Sizes))
clusterSizes

from pyspark.sql.functions import max, min

def cls_report(df):
  x1=df.select([min("x")])        # will return max value of each column
  x2=df.select([max("y")])
  x3=df.select([max("z")])
  return x1,x2,x3

#pick top out clusters with minimum instances
km_1st_cls = clusterSizes.values[0]
km_2nd_cls = clusterSizes.values[1]
km_3rd_cls = clusterSizes.values[2]
print(km_1st_cls)
print(km_2nd_cls)
print(km_3rd_cls)


#F1 = cls_report(pddf_pred.filter(f"prediction == {km_1st_cls}"))[0]
F1 = cls_report(pddf_pred.filter(f"prediction == {km_1st_cls}"))[0].select("min(x)").rdd.flatMap(list).collect()[0]
F2 = cls_report(pddf_pred.filter(f"prediction == {km_2nd_cls}"))[0].select("min(x)").rdd.flatMap(list).collect()[0]
F3 = cls_report(pddf_pred.filter(f"prediction == {km_3rd_cls}"))[0].select("min(x)").rdd.flatMap(list).collect()[0]

L1 = cls_report(pddf_pred.filter(f"prediction == {km_1st_cls}"))[1].select("max(y)").rdd.flatMap(list).collect()[0]
L2 = cls_report(pddf_pred.filter(f"prediction == {km_2nd_cls}"))[1].select("max(y)").rdd.flatMap(list).collect()[0]
L3 = cls_report(pddf_pred.filter(f"prediction == {km_3rd_cls}"))[1].select("max(y)").rdd.flatMap(list).collect()[0]

T1 = cls_report(pddf_pred.filter(f"prediction == {km_1st_cls}"))[2].select("max(z)").rdd.flatMap(list).collect()[0]
T2 = cls_report(pddf_pred.filter(f"prediction == {km_2nd_cls}"))[2].select("max(z)").rdd.flatMap(list).collect()[0]
T3 = cls_report(pddf_pred.filter(f"prediction == {km_3rd_cls}"))[2].select("max(z)").rdd.flatMap(list).collect()[0]

print(F1)
print(F2)
print(F3)

print(L1)
print(L2)
print(L3)

print(T1)
print(T2)
print(T3)


df_anomaly_1st_cls = pddf_pred.filter(f"(prediction == {km_1st_cls})") \
                                .filter(f"y == {L1}") \
                                .filter(f"z == {T1}") \
                                .filter(f"x == {F1}") 

display(df_anomaly_1st_cls)

我知道在 SciKit-learn 中的 KM 算法中，我们可以基于 post 使用：

clusters=KMeans(n_clusters=5)
df[clusters.labels_==0]

但是我们无法在 spark 中访问这样的 labels_ 来快速破解此任务。是否有任何优雅的方式（可能定义函数）来调用它，以便我们可以在主数据帧上反映任何聚类算法的结果以便更好地推理？

注意：我没有兴趣通过使用.toPandas() 将 spark 框架转换为 Pandas 数据农场来破解它

更新：我可能需要一个函数来根据多条件自动过滤根据实例和事件/行数获取输入数据帧和顶级集群的数量，在输出中返回过滤/选择的堆叠结果，其中预期数据帧的模拟如下：

def filtering(df, top_out_cluster=2, top_events/rows=2):

#                            +-----+----------+------------+----------+----------+
#                            |   id|prediction|           x|         y|         z|
#                            +-----+----------+------------+----------+----------+
#1st top out cluster         | row4|         3|   -5.122823|-3.3220499|-0.5069805|
#conditions F1, L1, T1       |row10|         3|  -5.6902847|-6.4827404|-0.9246967|
    
#2nd top out cluster         | row8|         7|  -0.2338185| -4.892681| 2.1228876|
#conditions F1, L1, T1       |row19|         7|  -2.0266335|-3.4171724|0.48218703|

#3rd top out cluster         |row18|         2|   3.3193955|-6.3573766| -6.978025|
#conditions F1, L1, T1       |row14|         2|   3.9688683|-6.0467844| -5.462389|
                            

#1st top out cluster         | row6|         5|   7.3153954| -5.079449| -7.291215|
#conditions F2, L2, T2       | row9|         5|    6.565442| -6.855994|-6.7983212|

#2nd top out cluster         |row12|         9|  -6.9042625|-6.1491723|-3.5354295|
#conditions F2, L2, T2       | row0|         9|  -6.0776997|-2.9096103|-1.5181729|

#1st top out cluster         | row1|         4|  -1.0122601|  7.322841|-5.4424076|
#conditions F3, L3, T3       |row11|         4|-0.017986143| 2.7632365| -8.814824|
                            
#2nd top out cluster         |row13|         1|  -10.389865|  9.537853|  0.674591|
#conditions F3, L3, T3       | row5|         1|  -2.4764006|  8.255791|  4.409478|
#                            +-----+----------+------------+----------+----------+

【问题讨论】：

目前尚不清楚您要达到什么目标以及您的预期输出是什么。此外，df_km_pred 是什么，你在哪里定义的？我错过了什么吗？另外，Freq、Length 和 Token_number 是在哪里创建的？
另外，“顶级”事件是什么意思？为什么某条记录“顶”？对于所有这些问题，我很抱歉，我正在努力理解您想要实现的目标
我更新了帖子，但我提供了colab notebook 以便快速调试。长话短说，在主数据框上分配聚类算法结果后，我想选择每个集群的前n行=事件并将它们堆叠并以数据框的形式报告。
感谢您的补充说明。我仍然不明白您所说的“顶部”是什么意思：确定每个已识别集群的“顶部”行的基本原理是什么？
感谢您的解决方案。它看起来不错，如果您以我在问题末尾的帖子更新中提到的函数形式形成您的解决方案，那就太好了。请在此处形成您的最终答案，以便我接受。

标签： machine-learning pyspark apache-spark-sql cluster-analysis hierarchical-clustering

【解决方案1】：

根据您在 colab 上讨论的要求，以下是代码。

from pyspark.sql.window import Window
from pyspark.sql import functions as f
from pyspark.sql.functions import row_number
winspec = Window.partitionBy("prediction").orderBy("prediction")

def get_top_n_clusters(model, top_out_cluster: int):
  n_clusters = model.summary.k
  cluster_size = model.summary.clusterSizes
  
  if top_out_cluster > n_clusters:
    raise ValueError(f"n value cannot be greater than cluster_size")

  col = ['size']
  cluster_size = pd.DataFrame(cluster_size, columns=col).sort_values(by=['size'], ascending=True)  #sorting
  return list(cluster_size.head(top_out_cluster).index)
  

def filtering(df, labels: list, top_records: int):
  winspec = Window.partitionBy("prediction").orderBy("prediction")
  return (
      df.filter(f.col("prediction").isin(labels))
        .withColumn("rowNum", f.row_number().over(winspec))
        .withColumn("minX", min(f.col("x")).over(winspec))
        .withColumn("maxY", max(f.col("y")).over(winspec))
        .withColumn("maxZ", max(f.col("z")).over(winspec))
        .filter(f.col("rowNum")<=top_records)
        .selectExpr("id", "prediction", "minX as x", "maxY as y", "maxZ as z")
  )

cluster_labels = get_top_n_clusters(model, top_out_cluster=4)
fdf = filtering(df_pred, labels=cluster_labels, top_records=1)
fdf.show()

+-----+----------+----------+----------+-----------+
|   id|prediction|         x|         y|          z|
+-----+----------+----------+----------+-----------+
|row15|         7|-10.505684|-1.6820424|-0.32242402|
| row4|         9| -9.426199| 0.5639291|  3.5664654|
| row0|         2| -8.317323|-1.3278837|-0.33906546|
|row14|         4|-0.9338185|-3.6411285| -3.7280529|
+-----+----------+----------+----------+-----------+

【讨论】：

我的解决方案中没有进行迭代。如果你可以分享详细的异常，我可以尝试修复它