如何在PySpark的Main函数中调用TeamCt与SeniorCt函数并获取结果

首先先修正你提供的函数中的语法错误（缺失闭合括号、智能引号问题）：

from pyspark.sql import functions as F

def TeamCt(source1):
    north_total = source1 \
        .filter(F.col('team') == 'North') \
        .groupBy(F.col('team')) \
        .agg(F.count('*').alias('n_total'))  # 补充缺失的闭合括号
    return north_total

def SeniorCt(source1):
    senior_total = source1 \
        .filter(F.col('grade') == 'Senior') \
        .groupBy(F.col('grade')) \
        .agg(F.count('*').alias('s_total'))  # 补充缺失的闭合括号
    return senior_total

接下来在Main函数中调用这两个函数，并根据需求整合结果：

方式1：提取统计值为变量并整合 #

如果只需要具体的统计数值，可以通过collect()提取结果后组合成字典等结构：

def Main(source1):
    # 调用两个统计函数
    north_stat_df = TeamCt(source1)
    senior_stat_df = SeniorCt(source1)
    
    # 提取统计值，同时处理空结果的情况
    north_count = north_stat_df.collect()[0]['n_total'] if north_stat_df.count() > 0 else 0
    senior_count = senior_stat_df.collect()[0]['s_total'] if senior_stat_df.count() > 0 else 0
    
    # 整合为统一结构返回
    combined_result = {
        "north_team_member_count": north_count,
        "senior_grade_member_count": senior_count
    }
    
    return combined_result

方式2：合并为单个DataFrame #

如果需要将两个统计结果合并成一个DataFrame，可以使用交叉连接或构造单行结构：

def Main(source1):
    north_stat_df = TeamCt(source1)
    senior_stat_df = SeniorCt(source1)
    
    # 交叉连接合并两个统计DataFrame
    combined_df = north_stat_df.crossJoin(senior_stat_df)
    
    # 可选：转为更简洁的单行结构
    combined_df = north_stat_df.select(F.lit("North").alias("team"), F.col("n_total")) \
        .crossJoin(senior_stat_df.select(F.lit("Senior").alias("grade"), F.col("s_total")))
    
    return combined_df

注意事项 #

• 确保source1是包含team和grade列的有效PySpark DataFrame
• 使用collect()时，因聚合后仅返回单行结果，不会引发内存问题，但需处理空结果避免索引越界
• 如果需要将结果写入存储或进一步分析，推荐使用DataFrame格式返回

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