基于配置文件重命名CSV列并存储至S3的PySpark/Python实现咨询

需求说明 #

现有存储在S3上的CSV文件表头含空格，源系统无法提前修改列名，需通过S3上的配置文件（记录源列名与目标列名的映射关系），将源CSV的列名替换为目标列名后重新存储到S3指定路径，要求支持任意列数的通用处理。

配置文件格式示例 #

src_column-name | target column-name
S.No | SNo
Count of lines visited | LinesVisited
Revenue | RevenueGenerated
No. of clicks | NumberofClicks

实现方案 #

方案一：PySpark（适合大文件分布式处理） #

1. 初始化Spark会话并读取配置文件

from pyspark.sql import SparkSession

# 初始化Spark会话
spark = SparkSession.builder.appName("CSVColumnRename").getOrCreate()

# 读取S3上的配置文件，替换为你的实际路径
config_df = spark.read.text("s3://your-config-bucket/config.txt")

# 解析配置文件，构建源列名到目标列名的映射字典
config_rdd = config_df.rdd \
    .filter(lambda line: not line.value.startswith("src_column-name")) \
    .map(lambda line: line.value.split("|")) \
    .map(lambda parts: (parts[0].strip(), parts[1].strip()))

column_mapping = dict(config_rdd.collect())

2. 读取源CSV并替换列名

# 读取S3上的源CSV文件，保留原始表头
source_df = spark.read.option("header", "true").csv("s3://your-source-bucket/source.csv")

# 遍历所有源列，用映射字典替换列名，无映射则保留原列名
renamed_df = source_df.select(
    [source_df[col].alias(column_mapping.get(col, col)) for col in source_df.columns]
)

3. 将处理后的数据写入S3目标路径

# 写入目标路径，覆盖已有文件，保留新表头
renamed_df.write.option("header", "true").mode("overwrite").csv("s3://your-target-bucket/renamed-data.csv")

方案二：纯Python（Pandas，适合小文件处理） #

import pandas as pd
import boto3

# 从S3下载配置文件到本地临时路径
s3 = boto3.client('s3')
s3.download_file("your-config-bucket", "config.txt", "/tmp/config.txt")

# 解析配置文件构建映射字典
column_mapping = {}
with open("/tmp/config.txt", "r") as f:
    next(f)  # 跳过表头行
    for line in f:
        line = line.strip()
        if not line:
            continue
        src_col, target_col = line.split("|")
        column_mapping[src_col.strip()] = target_col.strip()

# 直接读取S3上的源CSV文件
source_df = pd.read_csv("s3://your-source-bucket/source.csv")

# 替换列名
source_df.rename(columns=column_mapping, inplace=True)

# 将处理后的数据写入S3目标路径
source_df.to_csv("s3://your-target-bucket/renamed-data.csv", index=False)

关键注意事项 #

• 确保运行环境具备S3访问权限，可通过IAM角色或配置AWS密钥实现
• 若配置文件分隔符不是|，需对应调整代码中的split参数
• 大文件优先使用PySpark方案，避免内存溢出

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