PySpark中筛选列子集并对比忠诚与非忠诚客户销售额

嘿，我来帮你搞定这个PySpark里对比忠诚客户和非忠诚客户销售额的需求！下面是一步步的实现方案，代码和解释都给你安排明白：

1. 加载销售CSV数据 #

首先咱们得把数据读进Spark DataFrame里，记得指定header=True（如果你的CSV有表头的话），还可以根据实际情况调整分隔符、schema推断这些参数：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, when

# 初始化SparkSession
spark = SparkSession.builder.appName("LoyaltyCustomerSalesComparison").getOrCreate()

# 加载CSV文件
sales_df = spark.read.csv(
    path="/path/to/your/sales_data.csv",
    header=True,
    inferSchema=True,  # 自动推断列类型，也可以手动指定schema更严谨
    sep=","  # 如果是其他分隔符比如\t就改成对应值
)

2. 标记客户类型（核心逻辑） #

根据你给出的规则：collector_key为正整数是忠诚客户，负整数是非忠诚客户。咱们用when函数生成一个新列customer_type来区分两类客户，同时顺手处理可能的空值情况（比如把null归为非忠诚客户）：

# 添加客户类型标记列
sales_with_type_df = sales_df.withColumn(
    "customer_type",
    when(col("collector_key") > 0, "忠诚客户")
    .otherwise("非忠诚客户")  # 负整数和null都归为非忠诚范畴
)

3. 聚合计算两类客户的销售额 #

接下来按customer_type分组，聚合计算总销售额——你还可以根据需求加其他指标，比如订单数、平均销售额之类的：

# 分组计算总销售额
sales_comparison_df = sales_with_type_df.groupBy("customer_type")\
    .agg(
        sum("总销售额").alias("总销售额合计"),
        # 可选：添加其他聚合指标，比如订单数量
        # count("订单ID").alias("订单总数")
    )

4. 查看或输出结果 #

最后就可以直接查看对比结果，或者把结果保存成文件：

# 打印结果到控制台
sales_comparison_df.show()

# 保存结果为CSV（可选）
sales_comparison_df.write.csv(
    path="/path/to/output/comparison_result",
    header=True,
    mode="overwrite"  # 如果文件已存在就覆盖
)

额外小提示 #

• 如果你的collector_key列类型不是数值型，记得先转成整数：col("collector_key").cast("int")
• 如果需要更细致的对比（比如按门店、产品维度拆分），只需要在groupBy里添加对应的列即可，比如groupBy("customer_type", "门店ID")

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