Spark DataFrame分组内排除当前名称的统计量计算需求

我来帮你搞定这个需求！要实现每个org+name组合下，排除当前name后其余数据的均值、标准差和计数，我们可以分四步走，用PySpark的聚合和Join操作来实现：

步骤1：准备数据 #

首先先确认你的DataFrame创建代码（和你给出的一致）：

from pyspark.sql import functions as F

df = sqlContext.createDataFrame(
 [ ('org_1', 'a', 1), ('org_1', 'a', 2), ('org_1', 'a', 3), ('org_1', 'b', 4), ('org_1', 'c', 5), ('org_2', 'a', 7), ('org_2', 'd', 4), ('org_2', 'e', 5), ('org_2', 'e', 10) ],
 ["org", "name", "value"]
)

步骤2：计算每个org的整体统计量 #

我们需要先拿到每个org下所有数据的总和、平方和、总记录数，这是后续计算排除后统计量的基础：

org_stats = df.groupBy("org").agg(
    F.sum("value").alias("total_sum"),
    F.sum(F.pow("value", 2)).alias("total_sum_sq"),
    F.count("value").alias("total_count")
)

步骤3：计算每个org+name的分组统计量 #

接下来统计每个name在对应org内的总和、平方和、记录数，这样我们就知道要排除的那部分数据的情况：

name_org_stats = df.groupBy("org", "name").agg(
    F.sum("value").alias("name_sum"),
    F.sum(F.pow("value", 2)).alias("name_sum_sq"),
    F.count("value").alias("name_count")
)

步骤4：合并数据并计算排除后的统计量 #

把前两步的结果join起来，然后通过整体统计减去当前name的统计，得到排除后的指标，同时处理除数为0的边界情况（比如某个org只有一个name的所有记录）：

# 合并org级和org+name级的统计数据
joined_df = name_org_stats.join(org_stats, on="org", how="inner")

# 计算排除当前name后的各项指标
result_df = joined_df.withColumn("excluded_count", F.col("total_count") - F.col("name_count")) \
    .withColumn("excluded_sum", F.col("total_sum") - F.col("name_sum")) \
    # 计算均值，排除count为0的情况
    .withColumn("excluded_mean", 
                F.when(F.col("excluded_count") > 0, F.col("excluded_sum") / F.col("excluded_count")).otherwise(None)) \
    # 计算总体方差和标准差
    .withColumn("excluded_var_pop",
                F.when(F.col("excluded_count") > 0, 
                       (F.col("total_sum_sq") - F.col("name_sum_sq") - F.pow(F.col("excluded_sum"), 2)/F.col("excluded_count")) / F.col("excluded_count")).otherwise(None)) \
    .withColumn("excluded_stddev_pop", F.sqrt(F.col("excluded_var_pop"))) \
    # 计算样本方差和标准差（Spark默认的stddev是样本标准差）
    .withColumn("excluded_var_samp",
                F.when(F.col("excluded_count") > 1, 
                       (F.col("total_sum_sq") - F.col("name_sum_sq") - F.pow(F.col("excluded_sum"), 2)/F.col("excluded_count")) / (F.col("excluded_count") - 1)).otherwise(None)) \
    .withColumn("excluded_stddev_samp", F.sqrt(F.col("excluded_var_samp"))) \
    # 选择需要的输出字段
    .select("org", "name", "excluded_count", "excluded_mean", "excluded_stddev_pop", "excluded_stddev_samp")

查看结果 #

执行result_df.show()后，你会得到如下输出（保留两位小数方便查看）：

+-----+----+--------------+-------------+--------------------+--------------------+
|  org|name|excluded_count|excluded_mean|excluded_stddev_pop|excluded_stddev_samp|
+-----+----+--------------+-------------+--------------------+--------------------+
|org_1|   a|             2|          4.5|                 0.5| 0.7071067811865476|
|org_1|   b|             4|         2.75| 1.479019945774904| 1.707825127659933|
|org_1|   c|             4|          2.0| 1.118033988749895| 1.2909944487358056|
|org_2|   a|             3|         6.33| 3.0550504633038935| 3.5118845842842473|
|org_2|   d|             3|         7.33| 2.516611478423583| 3.0550504633038935|
|org_2|   e|             2|          5.5| 2.1213203435596424| 3.0|
+-----+----+--------------+-------------+--------------------+--------------------+

比如对于org_1的name=a，排除a后剩下b的4和c的5，计数是2，均值4.5，总体标准差0.5，完全符合你的预期~

内容的提问来源于stack exchange，提问作者Naomi Huang