PySpark读取不等列数文本文件生成DataFrame求助

核心思路 #

PySpark默认会依据第一行内容推断列数，导致后续行的多余字段被丢弃。要实现读取不定列数的文件，需先按行读取原始文本，再拆分每行字段并动态生成对应列。

具体实现步骤 #

• 按文本行读取文件
  使用textFile读取原始文件，得到每行的字符串数据：

  from pyspark.sql import SparkSession
  
  spark = SparkSession.builder.appName("ReadVariableColumns").getOrCreate()
  raw_data = spark.read.text("path/to/your/target_file.txt")
  

• 拆分每行字段
  用split函数按逗号拆分字段，同时处理字段前后的空格：

  from pyspark.sql.functions import split, trim, size
  
  split_data = raw_data.withColumn("fields", split(trim(raw_data.value), ",\\s*"))
  

  这里trim(raw_data.value)先清除每行首尾空格，",\\s*"匹配逗号加任意数量空格，确保拆分后的字段无冗余空格。

• 动态生成列
  先获取所有行中的最大字段数，再根据该值生成对应列：

  # 获取最大列数
  max_cols = split_data.select(size("fields")).agg({"size(fields)": "max"}).collect()[0][0]
  
  # 循环生成所有列
  for i in range(max_cols):
      split_data = split_data.withColumn(f"col_{i+1}", split_data.fields[i])
  
  # 移除临时的辅助列
  final_df = split_data.drop("value", "fields")
  

• 处理空值（可选）
  拆分后缺失的字段会显示为null，若需要替换为默认值，可使用fillna：

  final_df = final_df.fillna("")  # 用空字符串替换所有null值
  

效果验证 #

执行上述代码后，final_df的列数等于文件中最长行的字段数，不足列的位置会填充null（或指定的默认值），完美适配每行列数不同的场景。

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