Lambda函数无法将cities.csv数据写入DynamoDB weather表求助

我尝试用名为seedDynamo的Lambda函数，把cities.csv文件的数据导入到weather表，但写入操作一直卡在这条语句：

const response = await docClient.send(new PutCommand(params));

最终DynamoDB表还是空的，执行日志里只有"entered handler function"的信息，完全看不到写入相关的日志，求问题原因。

Lambda函数代码 #

// es6 syntax for including modules
import * as fs from 'fs';

// es6 syntax for including modules
import {DynamoDB, PutItemCommand, DynamoDBClient} from "@aws-sdk/client-dynamodb";

import { PutCommand, DynamoDBDocumentClient } from "@aws-sdk/lib-dynamodb";

export const handler = async(event, callback) => {

     console.log("entered handler function");

     const DDB = new DynamoDBClient({region: 'us-east-1'});
     const docClient = DynamoDBDocumentClient.from(DDB);
     let params = {};

     const fileContents = fs.readFileSync("cities.csv", "utf8").split('\n').map(async (item_str) => {
          params.ReturnConsumedCapacity = "TOTAL";
          params.TableName = "weather";
          params.Item = {
              "sc": {
                  "S": item_str.split(",")[0]
              },
              "t": {
                  "N": String(item_str.split(",")[1])
              }
          };
 //         console.log(params);
          try {
              const response = await docClient.send(new PutCommand(params));
              console.log(response);
          } catch (err) {
              console.log("Error", err);
          } finally {
              console.log("Success");
          }
      });
   
   setTimeout(function(){
       callback(null, "ok");
   }, 1000 * 10);
  
};

cities.csv部分记录 #

AGUADA,53
AKRON,22
ALTO,47
ANAHEIM,62
ANCHORAGE,57
ANN ARBOR,73
APPLE VALLEY,38
AUGUSTA,52
BARCELONETA,23
BERKELEY,60

执行日志 #

Test Event Name
seedDYnamoDB

Response
null

Function Logs
START RequestId: 71627f6e-f384-4f6d-aa81-95588d90710d Version: $LATEST
2023-06-07T13:42:00.121Z    71627f6e-f384-4f6d-aa81-95588d90710d    INFO    entered handler function
END RequestId: 71627f6e-f384-4f6d-aa81-95588d90710d
REPORT RequestId: 71627f6e-f384-4f6d-aa81-95588d90710d  Duration: 1574.02 ms    Billed Duration: 1575 ms    Memory Size: 128 MB Max Memory Used: 105 MB Init Duration: 506.99 ms

Request ID
71627f6e-f384-4f6d-aa81-95588d90710d

核心问题1：Lambda提前终止，未等待异步写入完成 #

你的handler是async函数，但你用了setTimeout调用callback，Lambda在callback执行后就会立刻结束，根本不会等待map里的异步Put操作完成。那些await docClient.send(...)还没执行，函数就已经终止了，自然不会有数据写入，也看不到相关日志。

核心问题2：共享参数对象引发竞态条件 #

你在map外面定义了let params = {}，然后每次迭代都修改这个对象。因为map里的回调是异步的，当第一个异步操作还在等待时，后面的迭代已经修改了params的内容，导致所有PutCommand最终都会用最后一次迭代的参数，甚至可能出现数据混乱。

核心问题3：文档客户端参数格式错误 #

你用的是DynamoDBDocumentClient（文档客户端），但给Item传的是低级API的格式（比如{"S": "AGUADA"}），文档客户端不需要指定类型，直接传原始值即可（比如"sc": "AGUADA"），格式错误会导致写入失败，但因为Lambda提前终止，你看不到错误日志。

次要问题：CSV空行处理不当 #

split('\n')会把文件末尾的空行也包含进去，导致item_str.split(",")[1]是undefined，转成字符串后是"undefined"，不符合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 () => {
    console.log("entered handler function");

    const DDB = new DynamoDBClient({ region: 'us-east-1' });
    const docClient = DynamoDBDocumentClient.from(DDB);

    // 读取CSV并过滤空行，避免处理无效数据
    const lines = fs.readFileSync("cities.csv", "utf8").split('\n').filter(line => line.trim() !== '');
    
    // 生成所有写入操作的Promise
    const writePromises = lines.map(async (item_str) => {
        const [sc, tStr] = item_str.split(",");
        const t = Number(tStr);
        
        // 跳过数值无效的行
        if (isNaN(t)) {
            console.log(`跳过无效行: ${item_str}`);
            return;
        }

        // 每次迭代创建新的params对象，避免竞态条件
        const params = {
            ReturnConsumedCapacity: "TOTAL",
            TableName: "weather",
            Item: {
                "sc": sc, // 文档客户端直接传字符串
                "t": t    // 直接传数值类型
            }
        };

        try {
            const response = await docClient.send(new PutCommand(params));
            console.log(`写入成功: ${sc}`, response);
        } catch (err) {
            console.log(`写入失败: ${sc}`, err);
        }
    });

    // 等待所有写入操作完成
    await Promise.all(writePromises);
    console.log("所有数据写入完成");
    return "ok";
};

修复要点： #

1. 移除callback和setTimeout，利用async函数的特性，Lambda会自动等待返回的Promise完成
2. 每次迭代创建独立的params对象，避免竞态条件
3. 使用文档客户端的正确参数格式，直接传原始值，不需要指定类型标记
4. 过滤CSV空行，处理无效数值，避免写入错误
5. 用Promise.all等待所有异步写入操作完成

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