Spark如何将含键值对的CSV文件解析为DataFrame（含缺失列场景）

针对你遇到的这种键值对格式、存在缺失列的平面文件，完全可以只用Spark Session完成解析，不需要依赖SparkContext。以下是具体步骤：

1. 读取文本文件为初始DataFrame #

首先直接用spark.read.text读取文件，得到包含每行内容的DataFrame：

from pyspark.sql import SparkSession
from pyspark.sql.functions import split, explode, regexp_extract, map_from_entries, array, struct

# 初始化Spark Session
spark = SparkSession.builder.appName("KeyValueFileParser").getOrCreate()

# 读取目标文件，生成每行对应一条记录的DataFrame
raw_df = spark.read.text("your_file_path.txt")

2. 提取所有可能的列名 #

因为文件中存在缺失列，首先需要收集所有出现过的列名，确保最终DataFrame包含全部字段：

# 将每行内容按竖线分割为键值对数组
raw_df = raw_df.withColumn("kv_array", split(raw_df["value"], "\\|"))

# 展开所有键值对，得到单独的键值记录
exploded_kv = raw_df.select(explode("kv_array").alias("kv_pair"))

# 从每个键值对中提取key和value
key_value_pairs = exploded_kv.select(
    regexp_extract("kv_pair", "^([^=]+)", 1).alias("column_name"),
    regexp_extract("kv_pair", "=(.*)$", 1).alias("value")
)

# 收集所有唯一的列名（全程使用Spark Session API，未依赖SparkContext）
all_columns = [row.column_name for row in key_value_pairs.select("column_name").distinct().collect()]

3. 将每行转换为结构化数据并补全缺失列 #

把每行的键值对数组转成Map类型，再基于收集到的所有列名提取对应值，缺失的列自动填充为null：

# 将每行的键值对数组转换为Map结构
structured_df = raw_df.withColumn(
    "kv_map",
    map_from_entries(
        array(
            *[struct(
                regexp_extract(pair, "^([^=]+)", 1).alias("key"),
                regexp_extract(pair, "=(.*)$", 1).alias("value")
            ) for pair in split(raw_df["value"], "\\|")]
        )
    )
)

# 提取所有列，缺失字段自动设为null
final_df = structured_df.select(
    *[structured_df["kv_map"].getItem(col).alias(col) for col in all_columns]
)

# 查看结果
final_df.show()

关键说明 #

• 为什么不用CSV读取：传统CSV按位置匹配列，但你的文件是键值对关联列名，缺失列直接不存在而非留空，位置无法对应列名，因此CSV方式不适用。
• 适配缺失列：通过先收集所有列名，再从Map中提取对应值的方式，不管缺失的列在哪个位置，都会自动填充null，完全符合需求。
• 无SparkContext依赖：全程使用spark.read及DataFrame API，避免了sc.textFile带来的Kerberos/Hadoop配置问题。

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