使用AWS Glue从S3导入数据到DynamoDB出现多余\字符问题

导入后出现的多余\是JSON多层转义导致的，核心问题出在字段映射逻辑：

• DynamoDB导出到S3的备份数据本身已经携带原生类型标记（S/M/N/BOOL/L等），是结构化的DynamoDB格式JSON，不是普通文本字符串
• 现有脚本通过ApplyMapping把Item.UserProgramId、Item.UserProgram两个字段强制转换为string类型，相当于把原本的结构化对象序列化成了带转义的JSON字符串，写入DynamoDB时连接器会自动给字符串类型值套一层"S"包装，最终就出现了值里嵌套带转义JSON的异常结构。

另外原脚本存在冗余代码：重复定义了两次S3数据源读取逻辑，其中S3bucket_node1创建后完全没有被使用，可以直接删除。

方案1：直接保留原生结构导入（推荐，适配全量备份恢复场景） #

Glue原生支持DynamoDB导出到S3的格式，不需要手动做字段类型映射，直接提取源数据里的Item结构写入即可，修复后的完整脚本：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# 读取S3上存储的DynamoDB备份数据
source_dyf = glueContext.create_dynamic_frame.from_catalog(
    database="dev-userprogram",
    table_name="data",
    transformation_ctx="source_dyf",
)

# 提取Item节点下的原生DynamoDB结构化数据，不做字符串转换
cleaned_dyf = source_dyf.map(lambda row: row["Item"])

# 写入目标DynamoDB表
glueContext.write_dynamic_frame_from_options(
    frame=cleaned_dyf,
    connection_type="dynamodb",
    connection_options={
        "dynamodb.region": "ap-southeast-2",
        "dynamodb.output.tableName": "Dev-UserProgram",
        "dynamodb.throughput.write.percent": "1.0"
    }
)

job.commit()

方案2：按需映射字段+反序列化（适配需要字段裁剪/转换的场景） #

如果不需要全量导入、必须做字段映射调整，先把读取到的字符串类型字段做JSON反序列化，解析成结构化对象后再写入，不要直接把序列化后的JSON字符串写入DynamoDB。
注意映射时不要把嵌套结构的字段强制指定为string类型，要对应DynamoDB的实际类型做映射，比如Map类型字段要指定为struct类型，避免触发自动序列化转义。

修复后重新执行作业，DynamoDB中存储的数据为原生格式，不存在多余转义符：

{
  "UserProgramId": {
    "S": "d135a9a8163d486d9398622e4301ab1b"
  },
  "UserProgram": {
    "M": {
      "EmailAddress": {"S": "test@gmail.com"},
      "EndDateUTC": {"S": "2021-02-12T11:12:55.543Z"},
      "FixedDuration": {"BOOL": true},
      "HasCompleted": {"BOOL": false},
      "PlanType": {"N": "1"},
      "RowVersion": {"S": "637516535873894162"},
      "ScheduleId": {"S": "4f4fe32cd8424cc190bbcfa3cdc8f2c1"},
      "StartDateRangeKey": {"S": "2021-03-18T08:39:47"},
      "StartDateUTC": {"S": "2021-03-18T08:39:47.3851194Z"},
      "UserProgramId": {"S": "d135a9a8163d486d9398622e4301ab1b"},
      "UserProgress": {"L": []},
      "__typename": {"S": "UserProgram"}
    }
  }
}

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