PySpark嵌套列DataFrame中如何将NONE转为空字符串？

问题背景 #

我有一个含嵌套列的PySpark DataFrame，结构定义如下：

df_schema = StructType([
    StructField("response", StringType(), True),
    StructField("id", StringType(), True),
    StructField("data", StructType([
      StructField("type", StringType(), True),
      StructField("record", StringType(), True),
      StructField("enteredCycle", StringType(), True),
      StructField("timestamp", StringType(), True),
      StructField("modifiedById", StringType(), True),
      StructField("years", IntegerType(), True),
      StructField("attributes", StructType([
        StructField("mass", ArrayType(DoubleType()), True),
        StructField("pace", ArrayType(IntegerType()), True),
        StructField("reflex", ArrayType(StringType()), True)
      ]))
    ]))
])

该DataFrame通过API调用生成，实现代码如下：

def api_call(parameter: str):
    response = session.get(f"https:url={parameter}", headers=header_data)
    return json.dumps(json.loads(response.text))

udf_call = udf(lambda z:api_call(z),StringType())

将该UDF添加到输入DataFrame生成api_response列，再通过from_json解析为指定Schema，合并结果后写入Parquet表时出现报错：

Job aborted due to stage failure: Task 24 in stage 161.0 failed 4 times, most recent failure: Lost task 24.3 in stage 161.0 (TID 1119) (100.66.2.91 executor 0): org.apache.spark.api.python.PythonException: 'TypeError: can only concatenate str (not "NoneType") to str', from year_loads.py, line 21.

推测是输出中的NONE值无法适配StringType，因此实现了将所有NONE转为空字符串的逻辑，但应用后仍报错：

# Get only String Columns
def replace_none_with_empty_str(df: DataFrame):
  string_fields = []
  for i, f in enumerate(df.schema.fields):
      if isinstance(f.dataType, StringType):
          string_fields.append(f.name)
  exprs = [none_as_blank(x).alias(x) if x in string_fields else x for x in df.columns]
  df.select(*exprs)
  return df

# NULL/NONE to blank logic
def none_as_blank(x):
    return when(col(x) != None, col(x)).otherwise('')


non_nulls_df = replace_none_with_empty_str(empty_df)
non_nulls_df.write.mode('overwrite').format('parquet').saveAsTable('dbname.tablename')

请问我的假设和解决方案是否正确？该逻辑是否覆盖了所有字符串列（尤其是嵌套字符串列）？若不正确，错误原因是什么？如何修正？

问题分析 #

1. 假设与解决方案的错误 #

• 假设错误：报错的直接原因不是DataFrame中的None值适配StringType，而是API调用UDF内部的字符串拼接错误——当输入的parameter为None时，f"https:url={parameter}"尝试拼接字符串与None，触发TypeError，这个错误发生在UDF执行阶段，远早于写入Parquet步骤。
• 解决方案错误：
  • 仅遍历顶层列，完全未处理data、data.attributes等嵌套结构中的字符串列；
  • 函数内df.select(*exprs)未赋值给变量，返回的仍是原DataFrame，等于未执行任何替换操作。

2. 错误根源拆解 #

报错指向year_loads.py第21行，对应API调用的字符串拼接代码。当输入参数为None时，Python不允许字符串与None直接拼接，直接抛出异常。后续的None替换逻辑既没触及这个根源问题，自身也存在功能缺陷。

修正方案 #

步骤1：修复API调用UDF的参数安全问题 #

在UDF内部先处理参数为None的情况，避免拼接报错：

def api_call(parameter: str):
    # 将None参数替换为空字符串或合法默认值
    safe_param = parameter if parameter is not None else ""
    response = session.get(f"https:url={safe_param}", headers=header_data)
    return json.dumps(json.loads(response.text))

udf_call = udf(lambda z: api_call(z), StringType())

步骤2：实现全层级字符串列的None替换逻辑 #

递归遍历所有嵌套结构，处理普通字符串列、嵌套Struct中的字符串列，以及字符串数组内的None值：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StringType, ArrayType

def replace_none_in_struct(struct_type: StructType, parent_col: str = ""):
    exprs = []
    for field in struct_type.fields:
        col_name = f"{parent_col}.{field.name}" if parent_col else field.name
        if isinstance(field.dataType, StructType):
            # 递归处理嵌套Struct
            nested_exprs = replace_none_in_struct(field.dataType, col_name)
            exprs.append(F.struct(*nested_exprs).alias(field.name))
        elif isinstance(field.dataType, ArrayType) and isinstance(field.dataType.elementType, StringType):
            # 将字符串数组内的None转为空字符串
            exprs.append(F.transform(F.col(col_name), lambda x: F.when(x.isNotNull(), x).otherwise("")).alias(field.name))
        elif isinstance(field.dataType, StringType):
            # 将普通字符串列的None转为空字符串
            exprs.append(F.when(F.col(col_name).isNotNull(), F.col(col_name)).otherwise("").alias(field.name))
        else:
            # 非字符串类型直接保留
            exprs.append(F.col(col_name).alias(field.name))
    return exprs

def replace_all_none_with_empty_str(df):
    all_exprs = replace_none_in_struct(df.schema)
    return df.select(*all_exprs)

步骤3：执行修正后的完整流程 #

# 生成API响应列（已修复参数安全问题）
df_with_api = input_df.withColumn("api_response", udf_call(F.col("your_parameter_col")))

# 解析为指定Schema
parsed_df = df_with_api.withColumn("parsed_data", F.from_json(F.col("api_response"), df_schema)).select("parsed_data.*")

# 全层级替换None为空字符串
cleaned_df = replace_all_none_with_empty_str(parsed_df)

# 写入Parquet表
cleaned_df.write.mode('overwrite').format('parquet').saveAsTable('dbname.tablename')

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