从动态深度嵌套的DataFrame中提取名为text的字符串值

针对层级不固定的Struct/Array混合深度嵌套结构，可通过递归遍历Schema的方式自动识别所有text字段路径，无需手动硬写嵌套层级，适配任意深度的动态嵌套结构。

实现步骤 #

• 递归遍历DataFrame的Schema结构，收集所有名为text的字段的完整访问路径
• 基于生成的路径批量提取字段，可按需合并为独立列或统一数组

完整可运行代码 #

from pyspark.sql.functions import col, flatten, array
from pyspark.sql.types import StructType, ArrayType, StringType

def find_all_text_fields(schema, parent_path=""):
    text_fields = []
    for field in schema.fields:
        # 拼接当前字段的完整访问路径
        current_path = f"{parent_path}.{field.name}" if parent_path else field.name
        field_type = field.dataType
        
        # 匹配到目标text字段直接存入路径列表
        if field.name == "text" and isinstance(field_type, StringType):
            text_fields.append(current_path)
        # Struct类型递归遍历内部字段
        elif isinstance(field_type, StructType):
            text_fields.extend(find_all_text_fields(field_type, current_path))
        # Array类型取元素结构递归，用[*]通配符匹配所有数组元素
        elif isinstance(field_type, ArrayType):
            element_type = field_type.elementType
            if isinstance(element_type, StructType):
                text_fields.extend(find_all_text_fields(element_type, f"{current_path}[*]"))
    return text_fields

# 自动获取所有text字段路径
text_paths = find_all_text_fields(df.schema)
# 方案1：每个text字段输出为独立列
result_df = df.select(*[col(path).alias(f"text_{idx}") for idx, path in enumerate(text_paths)])
# 方案2：所有text字段合并为一个一维数组输出
# result_df = df.select(flatten(array(*[col(path) for path in text_paths])).alias("all_texts"))

示例验证效果 #

针对你提供的测试数据，递归函数会自动识别到text字段路径为back.content[*].content[*].text，运行后可直接提取到目标字符串：

Vernetzte Komponenten innerhalb eines Netzwerks mit verschiedenen Hardware-Rechnern realisiert Ja.

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