使用PySpark DataFrame分组统计：区域维度客户来源计数需求

Got it, here's how you can achieve this with PySpark DataFrame step by step:

First, let's create the sample DataFrame from your provided customer data:

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

# Initialize Spark session
spark = SparkSession.builder.appName("CustomerSourceCount").getOrCreate()

# Sample customer data
data = [
    ("APAC", "mail", 1),
    ("APAC", "referral", 2),
    ("APAC", "mail", 3),
    ("APAC", "referral", 5),
    ("APAC", "mail", 6),
    ("APAC", "referral", 7),
    ("APAC", "referral", 8),
    ("US East", "mail", 9),
    ("US East", "referral", 10),
    ("US East", "walkIn", 11),
    ("AUS", "walkIn", 12),
    ("AUS", "referral", 13)
]

# Create DataFrame with specified columns
df = spark.createDataFrame(data, ["region", "source", "consumer_id"])

Next, we'll use grouping and pivoting to transform the data into your required format, handling missing sources and adjusting text cases as needed:

# Pivot source values into columns, count consumers, and format the output
result_df = df.groupBy("region") \
    # Explicitly list sources to ensure all required columns exist
    .pivot("source", ["mail", "referral", "walkIn"]) \
    .agg(count("consumer_id")) \
    # Fill empty source counts with 0
    .fillna(0) \
    # Rename columns to match your desired output
    .withColumnRenamed("mail", "mail_source_cnt") \
    .withColumnRenamed("referral", "referral_source_cnt") \
    .withColumnRenamed("walkIn", "walkIn_source_cnt") \
    # Adjust "US East" to uppercase as per your sample output
    .withColumn("region", when(col("region") == "US East", "US EAST").otherwise(col("region")))

# View the final result
result_df.show()

Breakdown of the steps: #

• groupBy("region"): Groups all records by their region to calculate per-region metrics.
• pivot("source", ["mail", "referral", "walkIn"]): Converts the source values into individual columns. By specifying the source list, we guarantee all three columns appear even if a region has no customers from a particular source.
• agg(count("consumer_id")): Counts the number of unique consumers for each source-region pair.
• fillna(0): Replaces null values (for sources with no customers in a region) with 0.
• Column renaming: Aligns the column names to match your requested output format.
• Region case adjustment: Converts "US East" to "US EAST" to match your sample result.

If you need to export the result to a CSV file, add this line:

result_df.write.csv("customer_source_summary.csv", header=True, mode="overwrite")

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