使用Databricks(PySpark)合并CSV：表头取自CSV1，数据取自CSV2

核心思路 #

先提取单独表头文件的列名，再将业务数据文件的默认列名替换为该表头，最终组合成目标DataFrame。

具体步骤 #

1. 读取表头文件并提取列名
   读取时禁用表头推断，将唯一一行数据作为列名列表提取：

   # 读取仅含表头的CSV，不将首行识别为表头
   header_df = spark.read.csv("/dbfs/path/to/your/header.csv", header=False, inferSchema=False)
   # 提取该行的所有值作为新列名
   target_columns = header_df.first().asDict().values()
   

2. 读取业务数据文件
   读取时可开启Schema推断（根据实际需求调整inferSchema参数），暂用默认列名：

   # 读取业务数据CSV，同样不使用内置表头
   data_df = spark.read.csv("/dbfs/path/to/your/data.csv", header=False, inferSchema=True)
   

3. 替换列名并生成最终DataFrame
   通过映射关系将默认列名替换为提取的表头：

   # 构建原列名到目标列名的映射
   rename_map = dict(zip(data_df.columns, target_columns))
   # 重命名列得到最终DataFrame
   final_df = data_df.select([col(old_name).alias(new_name) for old_name, new_name in rename_map.items()])
   

额外校验与优化 #

• 提前校验列数一致性，避免后续报错：

  assert len(data_df.columns) == len(target_columns), "表头文件与数据文件列数不匹配"
  

• 如果表头文件存在空行，先过滤无效行：

  header_df = header_df.filter(header_df._c0.isNotNull())
  

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