PySpark：将DataFrame所有字符串列截取至100字符写入Delta表

你的原代码逻辑完全错误，根本没触及到“截取列值”的需求，以下是问题分析和正确实现方式：

原代码的问题 #

• 逻辑偏离需求：你修改的是DataFrame的列名（field.name），但实际需要处理的是列的内容值，两者毫无关联。
• 性能风险极高：df.collect()[0]会把全量数据拉到Driver节点内存，面对大数据量时直接会引发内存溢出，完全没必要这么做。
• 空值处理不规范：PySpark中判断空值应该用is not None而非!= None，且你只判断了第一行的空值，无法覆盖所有行的情况。

正确实现方式 #

我们需要批量对所有字符串类型的列，将值截取为前100个字符（空值或长度≤100的内容保持不变），以下提供两种高效实现方式：

方式一：遍历字符串列逐列处理 #

from pyspark.sql.functions import col, substring, length, when

# 筛选出所有字符串类型的列名
string_columns = [field.name for field in df.schema.fields if field.dataType.typeName() == "string"]

# 对每个字符串列执行截取逻辑
for col_name in string_columns:
    df = df.withColumn(
        col_name,
        # 若字符串长度超过100则截取前100，否则保留原内容
        when(length(col(col_name)) > 100, substring(col(col_name), 1, 100)).otherwise(col(col_name))
    )

# 写入Delta表
df.write.format("delta").save("/your/delta/path")

方式二：一次性构造所有列（更高效） #

避免多次调用withColumn，直接通过select构造处理后的列列表，性能更优：

from pyspark.sql.functions import col, substring, length, when

# 区分字符串列和非字符串列
string_columns = [field.name for field in df.schema.fields if field.dataType.typeName() == "string"]
non_string_columns = [field.name for field in df.schema.fields if field.dataType.typeName() != "string"]

# 构造处理后的列：非字符串列直接保留，字符串列执行截取
processed_columns = non_string_columns + [
    when(length(col(c)) > 100, substring(col(c), 1, 100)).otherwise(col(c)).alias(c)
    for c in string_columns
]

# 生成处理后的DataFrame
df_processed = df.select(*processed_columns)

# 写入Delta表
df_processed.write.format("delta").save("/your/delta/path")

补充说明 #

• 上述逻辑会自动保留空值（length(null)返回null，otherwise分支会直接返回原空值）。
• 如果你的需求是“确保长度在1-100之间”（比如空值要填充默认字符），可以修改when逻辑，例如：

  when(length(col(col_name)) > 100, substring(col(col_name), 1, 100))
  .when(length(col(col_name)) < 1, lit("default_value"))  # 为空或长度为0时填充默认值
  .otherwise(col(col_name))
  

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