深入 PySpark：如何高效删除 DataFrame 中全为空的列

2026-02-17 23:12:38 0条评论 3次阅读 0人点赞

引言：处理数据中的“空洞”

在数据工程的日常工作中，我们经常面临着处理脏数据的挑战。特别是在使用 Apache Spark 进行大规模数据处理时，由于数据源的不一致性或 ETL 过程中的某些问题，我们经常会遇到包含大量空值（Null）的 DataFrame。这些全为空的列不仅占用宝贵的存储空间，还可能干扰后续的数据分析或机器学习模型训练。

你可能会遇到这样的情况：你从数据库加载了一张宽表，却发现其中某些列完全没有数据。或者，在进行了 Join 操作后，产生了一些在特定条件下全为 Null 的列。在这些情况下，为了保持数据集的整洁和计算的高效，我们需要找到一种方法，能够自动识别并删除这些“全空”的列。

在这篇文章中，我们将深入探讨如何使用 PySpark 来解决这一问题。我们将从基础的概念入手，逐步构建解决方案，并提供多种经过实战检验的代码示例。无论你是数据工程师还是数据科学家，掌握这一技巧都将帮助你编写更加健壮的数据处理脚本。

准备工作：创建示例环境

为了演示如何删除全空的列，我们首先需要一个包含此类数据的 PySpark DataFrame。让我们从创建一个 SparkSession 开始，并构建一个模拟数据集。

在下面的代码中，我们将构建一个包含员工信息的 DataFrame，其中特意包含了一列全为空的数据（middlename），以及包含部分空值和没有空值的数据。

from pyspark.sql import SparkSession
import pyspark.sql.types as T

# 创建 SparkSession
# 在实际生产环境中，通常会配置更多的参数，如 executor memory 等
spark = SparkSession.builder \
    .appName("RemoveNullColumns") \
    .getOrCreate()

# 定义数据
# 注意：第二列 "middlename" 全部为 None，这是我们接下来要移除的目标
actor_data = [
    ("James", None, "Bond", "M", 6000),
    ("Michael", None, None, "M", 4000),
    ("Robert", None, "Pattinson", "M", 4000),
    ("Natalie", None, "Portman", "F", 4000),
    ("Julia", None, "Roberts", "F", 1000)
]

# 定义 Schema
# 显式定义 Schema 是 Spark 开发的最佳实践，可以避免推断带来的性能开销
actor_schema = T.StructType([
    T.StructField("firstname", T.StringType(), True),
    T.StructField("middlename", T.StringType(), True),
    T.StructField("lastname", T.StringType(), True),
    T.StructField("gender", T.StringType(), True),
    T.StructField("salary", T.IntegerType(), True)
])

# 创建 DataFrame
df = spark.createDataFrame(data=actor_data, schema=actor_schema)

# 展示数据
print("原始 DataFrame:")
df.show(truncate=False)

输出结果：

原始 DataFrame:
+---------+----------+---------+------+------+
|firstname|middlename|lastname |gender|salary|
+---------+----------+---------+------+------+
|James    |null      |Bond     |M     |6000  |
|Michael  |null      |null     |M     |4000  |
|Robert   |null      |Pattinson|M     |4000  |
|Natalie  |null      |Portman  |F     |4000  |
|Julia    |null      |Roberts  |F     |1000  |
+---------+----------+---------+------+------+

方法一：基于字典映射的计数法（标准解法）

这是最常用且稳健的方法。其核心思想是：如果某一列中的 Null 值数量等于 DataFrame 的总行数，那么这一列就是全空的。

为了实现这一点，我们需要执行以下步骤：

  • 计算 DataFrame 的总行数。
  • 遍历所有列，计算每列中 Null 值的数量。
  • 将统计结果映射为一个字典或列表。
  • 对比 Null 数量与总行数，筛选出需要删除的列名。
  • 使用 drop 方法删除这些列。

让我们看看具体的代码实现：

import pyspark.sql.functions as F

# 第一步：获取 DataFrame 的总行数
# count() 会触发一个 Action，Job 会被提交并执行
df_size = df.count()
print(f"DataFrame 总行数: {df_size}")

# 第二步：计算每一列的空值数量
# 这里我们使用列表推导式配合 select 来一次性计算所有列的统计信息
# F.col(c).isNull() 返回布尔值，F.when 将其转换为 1 或 0
null_counts = df.select(
    [F.count(F.when(F.col(c).isNull(), c)).alias(c) for c in df.columns]
).collect()[0].asDict()

print(f"每列的空值统计: {null_counts}")

# 第三步：筛选出需要删除的列
# 如果某列的空值数量等于总行数，说明该列全为空
cols_to_drop = [k for k, v in null_counts.items() if v == df_size]
print(f"建议删除的列: {cols_to_drop}")

# 第四步：执行删除操作
# drop() 方法支持 *args，所以我们可以直接解包列表
final_df = df.drop(*cols_to_drop)

print("清洗后的 DataFrame:")
final_df.show(truncate=False)

输出结果：

DataFrame 总行数: 5
每列的空值统计: {‘firstname‘: 0, ‘middlename‘: 5, ‘lastname‘: 1, ‘gender‘: 0, ‘salary‘: 0}
建议删除的列: [‘middlename‘]
清洗后的 DataFrame:
+---------+---------+------+------+
|firstname|lastname |gender|salary|
+---------+---------+------+------+
|James    |Bond     |M     |6000  |
|Michael  |null     |M     |4000  |
|Robert   |Pattinson|M     |4000  |
|Natalie  |Portman  |F     |4000  |
|Julia    |Roberts  |F     |1000  |
+---------+---------+------+------+

深入解析：

这种方法的优点是逻辑非常清晰，易于调试。通过 INLINECODE3fa45a94，我们将统计结果拉取到了 Driver 端。由于 INLINECODE8d46b1bb 的数量通常不会是几十万级别的（这属于反模式），所以这部分内存开销是可以忽略不计的。

方法二：利用聚合函数优化性能（进阶技巧）

虽然第一种方法很直观，但在某些极度追求性能的场景下，我们可以尝试更“函数式”的写法。这种方法的核心是利用 Spark SQL 内部的聚合机制，一次性完成判断。

让我们创建一个更有挑战性的示例，其中包含多个全空列，并展示如何批量处理。

# 添加更多全空列以演示批量处理
# 假设我们有额外的无数据列：‘unused_col1‘, ‘unused_col2‘
from pyspark.sql import Row

# 使用 Row 对象和 add 方法动态添加列
data_with_extra_cols = [Row(**r.asDict(), **{"unused_col1": None, "unused_col2": None}) for r in df.collect()]

# 注意：这里只是为了演示数据构造，实际中很少这样 collect 后再创建
# 让我们直接生成一个新的 DataFrame
new_data = [
    ("James", None, "Bond", "M", 6000, None, None),
    ("Michael", None, None, "M", 4000, None, None),
    ("Robert", None, "Pattinson", "M", 4000, None, None)
]
new_schema = T.StructType([
    T.StructField("firstname", T.StringType(), True),
    T.StructField("middlename", T.StringType(), True),
    T.StructField("lastname", T.StringType(), True),
    T.StructField("gender", T.StringType(), True),
    T.StructField("salary", T.IntegerType(), True),
    T.StructField("unused_col1", T.StringType(), True),
    T.StructField("unused_col2", T.StringType(), True)
])

df_complex = spark.createDataFrame(data=new_data, schema=new_schema)
df_complex.show(truncate=False)

现在，让我们使用一种更简洁的聚合方式来删除这些列。我们可以计算每列的“非空值”数量。如果非空值数量为 0，则该列全为空。

# 使用 count(ignores nulls) 特性
# 对于大多数类型，count(col) 会忽略 null
# 如果结果为 0，则表示没有非空值

# 我们计算每一列的非空值总数
non_null_counts = df_complex.agg(
    *[F.count(c).alias(c) for c in df_complex.columns]
).collect()[0].asDict()

print(f"非空值统计: {non_null_counts}")

# 筛选出非空值数量为 0 的列
cols_to_drop_v2 = [k for k, v in non_null_counts.items() if v == 0]

print(f"将被删除的列: {cols_to_drop_v2}")

df_cleaned_v2 = df_complex.drop(*cols_to_drop_v2)
df_cleaned_v2.show(truncate=False)

为什么这种方法更好？

在某些 SQL 引擎实现中，原生聚合函数 INLINECODE43fd3a25 比 INLINECODE68f1de95 稍微更高效一些，因为它直接跳过了空值，而无需进行条件判断。这是一个微优化，但在处理数十亿行数据时，这些微小的优化会累积成显著的性能提升。

实战应用场景与最佳实践

#### 1. 自动化 ETL 管道中的数据清洗

在真实的 ETL（抽取、转换、加载）流程中，数据 Schema 可能会随着时间变化。例如，上游的数据库表结构修改了，增加了新列但旧数据中该列全为空。如果你在 Spark 作业中直接读取这些数据并写入 Hive 或 Parquet，可能会导致元数据不一致。

建议： 在将数据写入磁盘之前，封装一个通用的 drop_full_null_columns 函数。

def drop_null_columns(df):
    """
    删除 DataFrame 中所有全为 null 的列。
    这是一个通用的工具函数，可以嵌入到任何 PySpark 作业中。
    """
    # 获取总行数
    total_rows = df.count()
    
    # 计算每列的空值数
    # 使用 dict comprehension 使代码更紧凑
    null_counts = df.select(
        [F.count(F.when(F.col(c).isNull(), c)).alias(c) for c in df.columns]
    ).first().asDict()
    
    # 找出需要删除的列
    cols_to_drop = [c for c in df.columns if null_counts[c] == total_rows]
    
    if cols_to_drop:
        print(f"[INFO] 正在删除全空列: {cols_to_drop}")
        return df.drop(*cols_to_drop)
    
    return df

# 使用示例
cleaned_df = drop_null_columns(df_complex)
cleaned_df.show()

#### 2. 处理嵌套结构中的列

如果你的 DataFrame 包含嵌套的 StructType，上述方法主要针对顶层列。如果要处理深层嵌套字段，逻辑会复杂得多。通常情况下，我们建议先使用 explode 展开数组或者单独提取 Struct 中的字段进行扁平化处理，然后再应用上述清洗逻辑。保持数据的扁平化是 Spark 性能优化的关键原则之一。

#### 3. 性能优化建议

  • 缓存机制：如果在清洗数据后还需要多次使用该 DataFrame，并且数据量很大，建议在清洗后立即使用 df.cache()。清洗过程本身会扫描全表，这是一个昂贵的操作。缓存结果可以避免后续操作重复扫描数据。
    cleaned_df = drop_null_columns(df_large)
    cleaned_df.cache() # 将清洗后的数据缓存到内存中
    cleaned_df.count() # 触发缓存 Action
    # 后续操作将直接使用缓存数据
  • 并行度：在计算 null_counts 时，Spark 会自动并行处理。但如果你的 DataFrame 分区极少（例如只有 1 个分区），计算过程会变成串行。确保在读取数据时有合理的分区数（例如 200 个分区或集群 Core 总数的 2-3 倍）。

常见问题排查

Q: 如果我的 DataFrame 非常大，count() 操作非常慢怎么办？

A: INLINECODE7a5410df 确实是一个昂贵的 Action，因为它必须扫描所有数据。然而，为了判断一列是否“全”为空，我们必须知道总行数或者扫描全表。这是无法避免的代价。如果你只是想移除“大多数”为空的列（例如 99% 为空），你可以使用近似查询或采样，但这在 PySpark 中实现起来比较复杂且不精确。为了数据准确性，我们通常还是接受 INLINECODE0bbb1328 的开销。

Q: drop 方法会修改原始的 DataFrame 吗？

A: 不会。DataFrame 是不可变的。INLINECODE52972d25 返回了一个新的 DataFrame 引用。你必须将结果赋值给一个变量（如 INLINECODE065a3828），否则操作将不起作用。

总结

在这篇文章中，我们深入探讨了如何使用 PySpark 清理 DataFrame 中的全空列。我们不仅介绍了基于 INLINECODEc38ca511 计数和 INLINECODEab9cba62 聚合两种核心技术，还提供了封装好的实用函数供你在实际项目中直接使用。

关键要点回顾：

  • 识别原理：通过对比列中空值数量与 DataFrame 总行数来识别全空列。
  • 代码实现：利用 select 和列表推导式可以优雅地一次性统计所有列。
  • 性能考量：虽然计算需要扫描全表，但合理的分区和缓存策略可以最小化其影响。
  • 不可变性：记得将 drop 的结果重新赋值，因为 DataFrame 本身是不可变的。

通过掌握这些技术，你可以编写出更加自动化、健壮的 PySpark 数据处理脚本，从而专注于数据本身的价值，而不是被琐碎的脏数据问题所困扰。希望这些技巧对你有所帮助！

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。如需转载，请注明文章出处豆丁博客和来源网址。https://shluqu.cn/54352.html
点赞
0.00 平均评分 (0% 分数) - 0 票

相关文章美国1G带宽/1T流量高速vps $17.99/年