本文旨在解决使用aws sdk v3将csv文件数据批量写入dynamodb时遇到的常见问题，特别是数据写入不完整或操作挂起的情况。文章将重点讲解`dynamodbdocumentclient`的正确数据格式要求以及如何正确处理`async/await`与`Array.prototype.map`结合使用的异步操作，确保所有数据能够被成功写入。
在使用AWS Lambda函数或其他node.js环境将csv文件中的数据导入到Amazon DynamoDB时，开发者常会遇到数据写入操作无法完成，导致目标表为空或数据不完整的问题。这通常源于对AWS SDK v3中DynamoDBDocumentClient的数据格式理解偏差以及对javaScript异步编程模式（特别是async/await与Array.prototype.map结合使用）的处理不当。本教程将详细剖析这两个核心问题，并提供正确的解决方案。
问题一：DynamoDBDocumentClient 的数据格式要求
AWS SDK v3提供了两种主要的DynamoDB客户端：
- DynamoDBClient: 这是低级别的客户端，直接与DynamoDB API交互。在使用此客户端时，Item中的每个属性值都必须显式指定其DynamoDB类型，例如字符串类型为{ S: “value” }，数字类型为{ N: “123” }。
- DynamoDBDocumentClient: 这是高级别的文档客户端，它在DynamoDBClient之上提供了一个抽象层。它的主要优势在于能够自动处理javascript原生类型与DynamoDB类型之间的映射，极大地简化了数据操作。因此，在使用DynamoDBDocumentClient时，Item属性值应直接使用JavaScript原生类型（如字符串、数字、布尔值、对象、数组），而无需手动指定{ S: “…” }或{ N: “…” }等类型包装。
错误的Item格式示例 (当使用DynamoDBDocumentClient时):
params.Item = { "sc": { "S": item_str.split(",")[0] // 错误：文档客户端不需要 { "S": "..." } }, "t": { "N": String(item_str.split(",")[1]) // 错误：文档客户端不需要 { "N": "..." } } };
上述代码尝试为DynamoDBDocumentClient的PutCommand提供低级别客户端所需的类型包装。这会导致数据写入失败或行为异常，因为文档客户端期望的是纯JavaScript对象。
正确的Item格式示例 (当使用DynamoDBDocumentClient时):
params.Item = { "sc": item_str.split(",")[0], // 正确：直接使用字符串 "t": parseInt(item_str.split(",")[1]) // 正确：直接使用数字，确保类型为Number };
请注意，对于数字类型，我们应使用parseInt()或parseFloat()将字符串转换为实际的JavaScript数字类型，而不是仅仅使用String()包裹。
问题二：async/await与Array.prototype.map的异步操作处理
在JavaScript中，当Array.prototype.map的回调函数被声明为async时，map函数本身并不会等待每个异步操作完成。它会立即返回一个包含所有promise对象的数组，但这些Promise可能尚未解决（resolved）或拒绝（rejected）。如果父函数在这些Promise解决之前就提前退出，那么异步操作（如写入DynamoDB）将不会完成。
错误的异步处理示例：
const fileContents = fs.readFileSync("cities.csv", "utf8").split('n').map(async (item_str) => { // ... DynamoDB PutCommand 逻辑 ... const response = await docClient.send(new PutCommand(params)); // ... }); // 此时 fileContents 只是一个 Promise 数组，而不是最终结果 // 如果没有 await，函数可能会在此处或之后立即退出
在上述代码中，map函数返回了一个Promise数组，但这个数组本身没有被await。这意味着Lambda函数或其他执行环境可能会在所有PutCommand完成之前就认为执行完毕并退出，导致数据写入中断。
正确的异步处理示例：
为了确保所有异步写入操作都已完成，我们需要使用Promise.all()来等待所有由map生成的Promise都解决。
const fileContents = fs.readFileSync("cities.csv", "utf8").split('n'); const putPromises = fileContents.map(async (item_str) => { // ... // 在这里构建 params.Item，确保使用正确的文档客户端格式 params.Item = { "sc": item_str.split(",")[0], "t": parseInt(item_str.split(",")[1]) }; try { const response = await docClient.send(new PutCommand(params)); console.log("写入成功:", response); return response; // 返回每个 PutCommand 的结果 } catch (err) { console.error("写入错误:", err); throw err; // 抛出错误以便 Promise.all 可以捕获 } }); // 等待所有 Promise 完成 await Promise.all(putPromises); console.log("所有数据写入完成。");
通过await Promise.all(putPromises)，我们确保了父函数会一直等待，直到所有DynamoDB写入操作都成功完成或至少有一个操作失败。
综合解决方案示例
以下是一个集成了上述两种修正的完整Lambda函数示例，用于从csv文件批量导入数据到DynamoDB：
import * as fs from 'fs'; import { DynamoDBClient } from "@aws-sdk/client-dynamodb"; import { PutCommand, DynamoDBDocumentClient } from "@aws-sdk/lib-dynamodb"; export const handler = async (event, context) => { // 建议使用 context 而非 callback console.log("进入处理函数"); const DDB = new DynamoDBClient({ region: 'us-east-1' }); const docClient = DynamoDBDocumentClient.from(DDB); const tableName = "weather"; // 定义表名 let fileContents; try { // 假设 cities.csv 文件在 Lambda 部署包的根目录 fileContents = fs.readFileSync("/var/task/cities.csv", "utf8"); } catch (readErr) { console.error("读取CSV文件失败:", readErr); return { statusCode: 500, body: jsON.stringify({ message: "无法读取CSV文件" }) }; } const itemsToProcess = fileContents.split('n').filter(line => line.trim() !== ''); // 过滤空行 const putPromises = itemsToProcess.map(async (item_str) => { const parts = item_str.split(","); if (parts.length < 2) { console.warn(`跳过无效行: ${item_str}`); return null; // 返回 null 或抛出错误，根据需求处理 } const params = { TableName: tableName, Item: { "sc": parts[0].trim(), // 城市代码，直接字符串 "t": parseInt(parts[1].trim()) // 温度，直接数字 }, ReturnConsumedCapacity: "TOTAL" }; try { const response = await docClient.send(new PutCommand(params)); console.log(`成功写入项目: ${item_str}`, response.ConsumedCapacity); return response; } catch (err) { console.error(`写入项目失败: ${item_str}`, err); // 可以选择在这里重新抛出错误，或返回一个表示失败的对象 throw new Error(`Failed to write item ${item_str}: ${err.message}`); } }); try { // 等待所有写入操作完成 await Promise.all(putPromises); console.log("所有数据已成功写入DynamoDB。"); return { statusCode: 200, body: json.stringify({ message: "数据导入成功" }) }; } catch (allErr) { console.error("部分或全部数据写入失败:", allErr); return { statusCode: 500, body: JSON.stringify({ message: "数据导入失败", error: allErr.message }) }; } };
注意事项与最佳实践：
- 文件路径： 在AWS Lambda环境中，fs.readFileSync(“cities.csv”, “utf8”)默认会在/var/task/目录下查找文件。请确保你的CSV文件与Lambda函数的代码一同打包部署，并且路径正确。
- 错误处理： 务必在try…catch块中包裹异步操作，以便捕获和记录任何潜在的写入错误。Promise.all会在其中一个Promise被拒绝时立即拒绝，这有助于快速识别问题。
- Lambda超时： 如果要导入的数据量非常大，单个Lambda函数的执行时间可能会超过其配置的超时时间。考虑以下策略：
- 增加Lambda函数的内存和超时时间。
- 将CSV文件上传到S3，然后使用S3事件触发Lambda，并分批处理数据。
- 使用BatchWriteItemCommand进行批量写入，这比单个PutCommand更高效，但需要处理批量写入的限制（例如，每个请求最多25个项目）。
- callback与return： 对于async Lambda函数，推荐直接return一个Promise或一个async函数的结果，而不是使用callback。callback是旧版node.js Lambda运行时风格。
- 数据清理： 在处理CSV数据时，使用.trim()清理字符串中的空白字符是一个好习惯，可以避免因意外空格导致的问题。
通过遵循上述指导原则，开发者可以有效解决DynamoDB批量数据写入时遇到的常见问题，确保数据能够准确、完整地从CSV文件导入到DynamoDB表中。