PySpark Pandas DataFrame高效条件删行问题及filter失效排查

我正在处理一个PySpark Pandas DataFrame，示例数据如下：

| col1 |          col2        | col3 |
|------|----------------------|------|
| 1    |'C:\windows\a\folder1'|  3   |
| 2    |'C:\windows\a\folder2'|  4   |
| 3    |'C:\windows\b\folder1'|  4   |
| 4    |'C:\unix\b\folder2'   |  5   |
| 5    |'C:\unix\a\folder3'   |  4   |

实际数据集规模约5500万行，仅展示部分示例。我需要删除满足以下两个条件的行：

• col2为Windows服务器路径且包含文件夹'a'
• col3不等于3

预期结果如下：

| col1 |          col2        | col3 |
|------|----------------------|------|
| 1    |'C:\windows\a\folder1'|  3   |
| 3    |'C:\windows\b\folder1'|  4   |
| 4    |'C:\unix\b\folder2'   |  5   |
| 5    |'C:\unix\a\folder3'   |  4   |

注：行5保留因它是Unix服务器路径，行1保留因col3值为3。

由于ps.df.drop()无法基于索引删行，我尝试用ps.df.filter过滤冗余数据，但仅测试第一个条件时未找到任何结果：

df.filter(like='\\windows\\a\\', axis=0)

Out[1]:
    | col1  | col2  | col3  |
    |-------|-------|-------|
    | blank | blank | blank |

为验证DataFrame无问题，我执行以下代码能正确找到目标行：

df[ df.col2.str.contains('\\windows\\a\\', regex=False) ]

Out[2]:
    | col1 |          col2        | col3 |
    |------|----------------------|------|
    | 1    |'C:\windows\a\folder1'|  3   |

我也尝试了ps.sql()函数，结果与df.filter()一致。但df.col2.str.contains()在大数据集上效率较低，因此不想使用该方法。

请问为什么df.filter会失效？以及如何高效实现需求的行删除操作？

一、为什么df.filter会失效？ #

PySpark Pandas中的df.filter()方法（指定axis=0时）的作用是匹配行索引名称，而非针对列值进行模糊匹配。你传入的like='\\windows\\a\\'会被用来匹配行索引的标识，而不是col2列的内容，因此找不到符合条件的行，返回空结果。该方法的设计用途本就不是筛选列值包含特定字符串的行。

二、高效实现需求的行删除操作 #

针对5500万行的大数据集，推荐使用分布式向量化操作或Spark SQL优化查询，这两种方式的效率远高于纯Python级别的字符串匹配。

方法1：PySpark Pandas布尔索引（向量化） #

直接构造布尔条件，保留不符合删除规则的行：

# 构造保留条件：不满足「Windows路径含a且col3≠3」的行
keep_condition = ~(
    df.col2.str.contains('C:\\windows\\a\\', regex=False) & 
    (df.col3 != 3)
)
filtered_df = df[keep_condition]

这里的str.contains会被PySpark Pandas转换为底层Spark的向量化操作，性能远优于纯Python循环，适合大规模数据。

方法2：PySpark SQL查询 #

将DataFrame注册为临时视图，利用Spark SQL的查询优化器处理过滤逻辑：

# 注册临时视图
df.createOrReplaceTempView('path_table')

# 执行SQL过滤（注意SQL中反斜杠需双重转义）
filtered_df = ps.sql("""
    SELECT * 
    FROM path_table 
    WHERE NOT (col2 LIKE '%C:\\\\windows\\\\a\\\\%' AND col3 != 3)
""")

Spark SQL会自动生成最优执行计划，在分布式集群上的处理效率最高。

方法3：Spark原生API（推荐超大规模数据） #

如果可以切换到Spark DataFrame，直接使用Spark原生的filter方法：

# 转换为Spark DataFrame
spark_df = df.to_spark()
# 执行过滤
filtered_spark_df = spark_df.filter(
    ~(spark_df.col2.contains('C:\\windows\\a\\') & (spark_df.col3 != 3))
)
# 转换回PySpark Pandas DataFrame（若需保留原格式）
filtered_df = filtered_spark_df.to_pandas_on_spark()

Spark原生API的分布式处理能力对超大规模数据集的性能提升最为显著。

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