如何单次请求筛选DataFrame同类型日期列匹配阈值条件

现有一个包含约240个date、timestamp字段的大型DataFrame，字段允许为空，同时存在多层嵌套结构、数组嵌套结构。需要实现：不在请求中手动指定任何具体列名，仅通过单次计算找出所有存在日期/时间戳值小于/大于阈值X的列名。

示例数据与预期输出如下：

# 示例DataFrame
name: Evgenii, born: 1998, father: [name: Ivan, born: 1970], wtf: [foo: bar, omg: [2000, 2001, 2022]]
name: Ivan, born: 1970, father: [name: Vasilii, born: 1940]
name: Vasilii, born: 1940, father: [name: Pavel, born: 1917]

# 查找所有日期小于1950的列，预期输出：
born, father.born

# 查找所有日期小于1920的列，预期输出：
father.born

# 查找所有日期大于1990的列，预期输出：
born, wtf.omg

（要求请求逻辑中不允许硬编码列名）

核心思路是自动递归遍历DataFrame全量Schema识别所有日期/时间戳字段，自动生成跨嵌套结构、数组的判断逻辑，单次作业完成扫描，适配空值、任意层级嵌套、数组嵌套的场景。

• 递归遍历Schema，提取所有DateType/TimestampType类型字段的完整点分隔路径（比如father.born、wtf.omg），自动处理struct嵌套、array嵌套场景
• 针对每个字段路径自动生成取值逻辑：普通字段直接取、struct嵌套按路径逐层取、array类型字段用数组存在判断函数检测是否有元素满足阈值
• 聚合所有字段的判断表达式，单次触发计算，返回所有存在满足阈值的记录的字段名

可直接运行的PySpark实现代码 #

from pyspark.sql import functions as F
from pyspark.sql.types import *
from typing import List, Tuple

# 递归提取所有日期/时间戳字段的路径和对应判断表达式
def extract_datetime_fields(schema: StructType, parent_path: str = "", threshold_val, cmp_op: str = "lt") -> List[Tuple[str, F.Column]]:
    field_exprs = []
    for field in schema.fields:
        current_path = f"{parent_path}.{field.name}" if parent_path else field.name
        data_type = field.dataType

        # 处理数组类型：判断数组内是否存在满足阈值的元素
        if isinstance(data_type, ArrayType):
            element_type = data_type.elementType
            # 数组元素是基本日期/时间戳类型
            if isinstance(element_type, (DateType, TimestampType)):
                if cmp_op == "lt":
                    cond = F.exists(F.col(current_path), lambda x: x < threshold_val)
                else:
                    cond = F.exists(F.col(current_path), lambda x: x > threshold_val)
                field_exprs.append((current_path, cond))
            # 数组元素是嵌套struct，继续递归
            elif isinstance(element_type, StructType):
                field_exprs.extend(extract_datetime_fields(element_type, current_path, threshold_val, cmp_op))
            # 数组元素还是数组，继续递归（适配多维数组）
            elif isinstance(element_type, ArrayType):
                field_exprs.extend(extract_datetime_fields(StructType([StructField("val", element_type)]), current_path, threshold_val, cmp_op))
        
        # 处理struct类型：继续递归遍历子字段
        elif isinstance(data_type, StructType):
            field_exprs.extend(extract_datetime_fields(data_type, current_path, threshold_val, cmp_op))
        
        # 处理普通日期/时间戳字段
        elif isinstance(data_type, (DateType, TimestampType, IntegerType)):
            if cmp_op == "lt":
                cond = F.col(current_path) < threshold_val
            else:
                cond = F.col(current_path) > threshold_val
            field_exprs.append((current_path, cond))
    return field_exprs

# 核心方法：传入df、阈值、比较运算符（lt=小于，gt=大于），返回符合条件的列名
def find_cols_with_threshold(df, threshold, cmp_op="lt"):
    # 提取所有日期字段的判断表达式
    field_conds = extract_datetime_fields(df.schema, threshold_val=threshold, cmp_op=cmp_op)
    if not field_conds:
        return []
    # 构造聚合表达式：只要任意一条记录满足条件，就标记该字段为True
    agg_exprs = [F.max(cond.cast("int")).alias(col_name) for col_name, cond in field_conds]
    # 单次触发计算，拿到结果
    res_row = df.agg(*agg_exprs).first()
    # 筛选出满足条件的字段
    return [col_name for col_name, _ in field_conds if res_row[col_name] == 1]


# ---------------------- 测试验证 ----------------------
# 构造示例数据
data = [
    ("Evgenii", 1998, ("Ivan", 1970), ("bar", [2000,2001,2022])),
    ("Ivan", 1970, ("Vasilii", 1940), None),
    ("Vasilii", 1940, ("Pavel", 1917), None)
]
schema = StructType([
    StructField("name", StringType()),
    StructField("born", IntegerType()),
    StructField("father", StructType([
        StructField("name", StringType()),
        StructField("born", IntegerType())
    ])),
    StructField("wtf", StructType([
        StructField("foo", StringType()),
        StructField("omg", ArrayType(IntegerType()))
    ]))
])
df = spark.createDataFrame(data, schema=schema)

# 测试1：找所有日期小于1950的列
print(find_cols_with_threshold(df, 1950, "lt")) # 输出 ['born', 'father.born']
# 测试2：找所有日期小于1920的列
print(find_cols_with_threshold(df, 1920, "lt")) # 输出 ['father.born']
# 测试3：找所有日期大于1990的列
print(find_cols_with_threshold(df, 1990, "gt")) # 输出 ['born', 'wtf.omg']

注：如果实际业务里日期字段是标准DateType/TimestampType类型，移除代码类型判断里的IntegerType即可，示例中保留是因为测试数据用整数年份模拟日期值。代码自动兼容空值、任意层级struct嵌套、数组/多维数组嵌套场景，全程不需要手动传入列名，全表仅触发一次计算，能直接适配百级以上字段的大型DataFrame扫描需求。

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