使用AWS DMS从CSV（S3）到DynamoDB的数据迁移

Question

当我迁移一张 json 的表格时，我正面临一个问题，该表应将其插入为 dyanmodb 地图。源数据采用以下格式

<table border="1"><tr><th>userId</th><th>brandId</th><th>createdBy</th><th>modifiedBy</th><th>preferences</th><th>version</th></tr><tr><td>TU0001</td><td>TEST BRAND</td><td>{"channel":{"S":"website"},"date":{"S": "2022-06-13T08:16:26.300Z"},"userId":{"S":"TU0001"}}</td><td>{"channel":{"S":"website"},"date":{"S": "2022-06-13T015:26:10.200Z"},"userId":{"S":"TU0001"}}</td><td>{"Colour": {"S": "Red" },"Size":{"S": "XL" }}</td><td>1</td></tr></table>

DynamoDB中的表位于SAM结构中，只是将JSON值存储为MAP（DynamoDB-MAP）。当我们使用DMS迁移时，它将JSON作为字符串值插入，而不是按预期映射。

我已经定义了转换规则如下：

"attribute-mappings": [
      {
        "target-attribute-name": "userId",
        "attribute-type": "scalar",
        "attribute-sub-type": "string",
        "value": "${userId}"
      },
      {
        "target-attribute-name": "brandId",
        "attribute-type": "scalar",
        "attribute-sub-type": "string",
        "value": "${brandId}"
      },
      {
        "target-attribute-name": "createdBy",
        "attribute-type": "document",
        "attribute-sub-type": "dynamodb-map",
        "value": {
          "M": {
            "S": "${createdBy}"
          }
        }
      },
      {
        "target-attribute-name": "modifiedBy",
        "attribute-type": "document",
        "attribute-sub-type": "dynamodb-map",
        "value": {
          "M": {
            "S": "${modifiedBy}"
          }
        }
      },
      {
        "target-attribute-name": "preferences",
        "attribute-type": "document",
        "attribute-sub-type": "dynamodb-map",
        "value": {
          "M": {
            "S": "${preferences}"
          }
        }
      },
      {
        "target-attribute-name": "version",
        "attribute-type": "scalar",
        "attribute-sub-type": "number",
        "value": "${version}"
      }
    ]

我还尝试在下面添加地图，并在Dyanmodb中添加一个空的地图。

"value": {
          "M": "${preferences}"
        }

希望有人可以帮忙。

Answer 1

我认为DMS不支持将JSON字符串直接转换为发电机图。在这里，DMS将JSON数据插入字符串，而不是转换为DynamoDB地图或列表类型。但是，您可以在迁移后或之前处理数据，以确保将JSON数据正确解析到Dyanamodb映射中。

迁移之前：

• 源数据库中的过程数据将JSON字符串转换为一种格式，该格式
  DM可以处理。
• 编写脚本（JS/Python）以读取您的数据
  源，将JSON转换为DynamoDB映射，然后写入DynamoDB。

---

或

迁移后：

• 读取来自dynamodb的数据，然后将JSON字符串放入映射，然后
  然后写回DynamoDB。
• 编写一个脚本，该脚本每次在dynamodb表中每次都在
  检索字符串数据，将其转换为地图，然后更新项目
  在DynamoDB中。

以下是一个node.js，具有AWS SDK的“后处理”示例，可以在迁移后完成，您可以做：

const AWS = require('aws-sdk');

AWS.config.update({
  region: "your_region",
  // Add your credentials
});

const dynamodb = new AWS.DynamoDB.DocumentClient();
const tableName = 'your_table'; // replace with your table name

const scanAndProcessItems = async (startKey) => {
  try {
    const params = {
      TableName: tableName,
      ExclusiveStartKey: startKey
    };
    
    const data = await dynamodb.scan(params).promise();
    
    if (data.Items) {
      for (let item of data.Items) {
        // To parse JSON strings into DynamoDB Maps
        item.createdBy = JSON.parse(item.createdBy);
        item.modifiedBy = JSON.parse(item.modifiedBy);
        item.preferences = JSON.parse(item.preferences);
    
        // To write the item back to the table
        await dynamodb.put({ TableName: tableName, Item: item }).promise();
      }
    }
    
    // If there's more data to scan, it will recursively call this function
    if (data.LastEvaluatedKey) {
      await scanAndProcessItems(data.LastEvaluatedKey);
    }
    
  } catch (err) {
    console.error("An error occurred", err);
  }
};

scanAndProcessItems();

希望这对您有帮助。