基于PySpark构建动态数据概况报告函数的需求

以下是一个动态生成数据概况报告的PySpark函数，包含基础统计、唯一值分析及可视化内容：

1. 函数实现 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, countDistinct, desc
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

def generate_data_profile_report(df, top_n_freq=3):
    # 1. 基础统计信息（基于summary扩展）
    basic_stats = df.summary().toPandas()
    
    # 2. 唯一值统计信息
    unique_stats = []
    for column in df.columns:
        # 计算基础指标
        col_counts = df.select(
            count(col(column)).alias("total_count"),
            countDistinct(col(column)).alias("distinct_count"),
            (countDistinct(col(column))/count(col(column))).alias("distinct_ratio")
        ).toPandas().iloc[0]
        
        # 计算Top N频繁值
        top_values = df.groupBy(column).count().orderBy(desc("count")).limit(top_n_freq).toPandas()
        top_values.columns = [f"{column}_value", "frequency"]
        
        # 整理当前列的唯一值统计
        col_stats = {
            "column": column,
            "total_count": col_counts["total_count"],
            "distinct_count": col_counts["distinct_count"],
            "distinct_ratio": round(col_counts["distinct_ratio"], 4),
            "top_frequent_values": top_values.to_dict("records")
        }
        unique_stats.append(col_stats)
    
    unique_stats_df = pd.DataFrame(unique_stats)
    
    # 3. 可视化生成
    plt.figure(figsize=(15, 8))
    
    # 子图1：各列唯一值数量柱状图
    plt.subplot(1, 2, 1)
    sns.barplot(x="column", y="distinct_count", data=unique_stats_df)
    plt.title("各列唯一值数量")
    plt.xticks(rotation=45)
    
    # 子图2：各列唯一值占比柱状图
    plt.subplot(1, 2, 2)
    sns.barplot(x="column", y="distinct_ratio", data=unique_stats_df)
    plt.title("各列唯一值占比")
    plt.xticks(rotation=45)
    
    plt.tight_layout()
    
    # 返回报告内容
    return {
        "basic_statistics": basic_stats,
        "unique_value_statistics": unique_stats_df,
        "visualization": plt
    }

2. 使用示例 #

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

# 创建示例DataFrame
data = [
    ("Alice", 25, "F", "New York"),
    ("Bob", 30, "M", "London"),
    ("Alice", 25, "F", "New York"),
    ("Charlie", 35, "M", "Paris"),
    ("Bob", 30, "M", "London")
]
df = spark.createDataFrame(data, ["name", "age", "gender", "city"])

# 生成报告
report = generate_data_profile_report(df)

# 打印基础统计
print("=== 基础统计信息 ===")
print(report["basic_statistics"])

# 打印唯一值统计
print("\n=== 唯一值统计信息 ===")
print(report["unique_value_statistics"])

# 展示可视化图表
report["visualization"].show()

函数说明 #

• 基础统计：基于PySpark的summary()函数，包含计数、均值、标准差、最小值、分位数及最大值
• 唯一值统计：对每列计算总条数、唯一值数量、唯一值占比，以及指定数量的高频值
• 可视化：生成两个柱状图，分别展示各列的唯一值数量和占比，帮助快速识别高基数/低基数列

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