如何在AWS上通过MapReduce方式将S3的GZIPPED CSV转为DynamoDB备份格式

你提到的需求完全可以通过AWS Data Pipeline + EMRActivity的方案落地，也可以直接用EMR的Spark任务完成S3上GZIP压缩CSV到DynamoDB原生JSON备份格式的转换，以下是具体的落地方案：

Spark原生支持GZIP等压缩格式的自动识别和解压，不需要额外处理GZIP源文件，只需要按如下步骤操作即可：

• 提前配置好EMR集群的访问权限：确保EMR运行角色拥有S3源路径读权限、S3输出路径写权限
• 编写PySpark转换脚本，逻辑完全匹配你的字段映射要求，示例代码如下：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_json, struct

if __name__ == "__main__":
    spark = SparkSession.builder.appName("CSV2DynamoBackup").getOrCreate()
    # 读取S3上的GZIP CSV，header=True表示读取第一行作为字段名
    df = spark.read.csv("s3://你的输入桶路径/*.csv.gz", header=True, inferSchema=False)
    # 字段映射：将原date字段重命名为uploadDate，按DynamoDB要求封装为字符串类型结构
    dynamo_df = df.select(
        struct(col("id").alias("s")).alias("id"),
        struct(col("name").alias("s")).alias("name"),
        struct(col("date").alias("s")).alias("uploadDate")
    )
    # 输出为每行一条JSON的格式，即DynamoDB导入要求的标准备份格式
    dynamo_df.select(to_json(struct("*")).alias("value")).write.mode("overwrite").text("s3://你的输出桶路径/dynamo_backup/")
    spark.stop()

• 在EMR控制台提交Spark步骤，选择上述脚本即可运行，运行完成后输出路径的文件就是你需要的目标格式，后续可以直接用aws dynamodb import-table命令批量导入到DynamoDB表。

如果需要定期执行转换任务，可以用Data Pipeline做调度，步骤如下：

• 打开Data Pipeline控制台新建管道，配置源数据位置为S3上的GZIP CSV路径，输出位置为备份文件存储的S3路径
• 新增EMRActivity节点，核心配置如下：
  • EMR集群可选择按需临时创建，或复用已有的EMR集群
  • 步骤字段填入Spark提交命令：spark-submit s3://你的脚本存储路径/csv2dynamo.py
  • 配置运行角色权限，确保DataPipeline默认角色和EMR角色拥有S3、EMR的相关操作权限
• 配置调度规则：可设置为单次运行，也可以设置定时触发适配定期的CSV同步需求
• 保存并激活管道即可自动运行，运行日志可以在Data Pipeline控制台直接查看。

补充说明：如果后续你的CSV字段包含数字、布尔等非字符串类型，只需要把对应字段封装时的s替换为n（数字）、bool（布尔值）即可，无需修改整体转换逻辑。

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