PySpark：如何从DataFrame中提取所有日期列生成子集

要从包含大量列的PySpark DataFrame中提取仅含日期值的列，可通过以下步骤实现：

1. 核心思路 #

由于PySpark是分布式计算框架，不能直接将全量数据拉到Driver端用dateutil.parser遍历验证。需通过**UDF（用户自定义函数）**结合列统计的方式，判断每列是否符合日期列的特征，再筛选目标列。

2. 具体实现代码 #

导入依赖库

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col, count, when
from pyspark.sql.types import BooleanType
from dateutil.parser import parse
from dateutil.parser._parser import ParserError

初始化SparkSession并创建示例数据

spark = SparkSession.builder.appName("DateColumnFilter").getOrCreate()

# 模拟用户提供的原始数据
data = [
    (1, 'John Smith', '12/10/1982', '123 Main St', '01/01/2000'),
    (2, 'Jane Smith', '11/21/1999', 'Abc St', '12/12/2020')
]
df = spark.createDataFrame(data, ["id", "name", "dob", "address", "join_date"])

定义日期验证UDF

该函数尝试用dateutil.parser解析输入值，能处理多种日期格式，解析成功返回True，失败返回False：

def is_date(s):
    try:
        parse(s)
        return True
    except (ParserError, TypeError, ValueError):
        return False

is_date_udf = udf(is_date, BooleanType())

统计每列的有效日期数量并筛选日期列

# 获取总记录数，用于计算比例
total_rows = df.count()

# 统计每列中可被解析为日期的记录数
date_count_df = df.select([
    count(when(is_date_udf(col(col_name)) == True, col_name)).alias(col_name)
    for col_name in df.columns
])

# 将统计结果转为字典，方便筛选
count_dict = date_count_df.collect()[0].asDict()

# 筛选出所有值都是日期的列（可根据需求调整阈值，比如允许10%以内的非日期值）
date_columns = [col_name for col_name, cnt in count_dict.items() if cnt == total_rows]

生成仅含日期列的新DataFrame

date_only_df = df.select(date_columns)
date_only_df.show()

3. 注意事项 #

• 阈值调整：如果列中存在少量非日期值，可将筛选条件改为cnt / total_rows >= 0.9（保留90%以上是日期的列）。
• 避免误判：dateutil.parser可能会将部分非日期字符串（如"1234"）误判为日期，若需要更严格的验证，可结合正则表达式或指定目标日期格式（如%m/%d/%Y）来优化UDF。
• 性能优化：对于超大规模数据集，count()操作会触发全量计算，可采用采样数据（如df.sample(0.1).count()）来快速判断列类型，减少计算开销。

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