PySpark循环查询Hive多分组数据并动态存储为DataFrame

需要对Hive中存储的表按group字段的80个分组值循环查询，将每个分组的结果转为独立的数据集，手动逐次修改参数重复执行代码效率极低，原有手写迭代代码存在多处逻辑和语法错误无法正常运行。

• 预定义df_list时引用了尚未创建的group_1到group_80变量，运行时会直接触发变量未定义错误
• SQL语句中使用$i做参数占位不符合PySpark的字符串语法规则，分组ID参数无法正常传入查询语句
• 嵌套循环逻辑完全冗余：外层遍历不存在的df_list、内层遍历分组ID，会导致无意义的重复执行
• 顶层脚本中错误使用return关键字，该关键字仅能在函数体内使用，直接调用会抛出语法错误
• range(1,80)为左闭右开区间，仅会生成1-79的整数，漏掉第80组的查询
• 循环内没有对结果做持久化存储，每次迭代生成的group_i变量会被下一次循环覆盖，最终无法获取全部分组结果

优先使用字典统一存储所有分组结果，键为分组ID、值为对应分组的Pandas DataFrame，既方便按分组ID索引调用，也不会产生大量零散全局变量污染命名空间

# 初始化字典存储全部分组结果
group_dfs = {}

# 遍历1-80所有分组
for group_id in range(1, 81):
    # 拼接查询SQL，注意group是SQL保留字，用反引号包裹避免解析报错
    query_sql = f"""
        SELECT * FROM db.hive_table WHERE `group` = {group_id}
    """
    # 执行Spark SQL查询
    spark_temp = spk.sql(query_sql)

    # 打印校验信息（和原有手动执行的校验逻辑一致）
    print(f"===== 分组 {group_id} 数据预览 =====")
    spark_temp.show(3)
    spark_temp.printSchema()
    print(f"分区数据行数: {spark_temp.count()}, 表字段数: {len(spark_temp.columns)}")

    # 转为Pandas DataFrame存入字典
    group_dfs[group_id] = spark_temp.toPandas()

后续需要调用指定分组的数据时，直接按key取值即可，比如要取group=1的结果，直接写group_dfs[1]。

如果确实需要生成group_1、group_2这类独立命名的变量（不推荐，变量太多不好维护），可以通过动态注册全局变量实现：

for group_id in range(1, 81):
    query_sql = f"""
        SELECT * FROM db.hive_table WHERE `group` = {group_id}
    """
    spark_temp = spk.sql(query_sql)
    # 此处省略和上面一致的打印校验逻辑
    # 动态注册全局变量
    globals()[f"group_{group_id}"] = spark_temp.toPandas()

• 如果单分组数据量较大，不建议直接全量转为Pandas DataFrame，Pandas运行在驱动节点单机内存中，数据量过大会直接触发OOM内存溢出
• 逐组发起SQL查询的效率很低，如果后续对每个分组的处理逻辑一致，建议一次性加载全表到Spark DataFrame，通过groupBy算子或者按分区遍历的方式做处理，比循环执行80次查询性能高很多
• 写Spark SQL时注意规避保留字作为字段名的问题，遇到保留字字段务必加反引号包裹，避免不必要的语法报错

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