如何通过Node.js获取DynamoDB表标签（基于AWS SDK）

以下是基于@aws-sdk/client-dynamodb实现的完整方案，可在Lambda中动态监听DynamoDB表指定字段的变更：

1. 安装必要依赖 #

需要安装AWS SDK的DynamoDB客户端及数据转换工具：

npm install @aws-sdk/client-dynamodb @aws-sdk/util-dynamodb

2. 编写获取监听字段的工具函数 #

通过ListTagsOfResource API读取表的标签，解析出field_names对应的字段列表：

const { DynamoDBClient, ListTagsOfResourceCommand } = require("@aws-sdk/client-dynamodb");

// 初始化DynamoDB客户端
const dynamoClient = new DynamoDBClient({ region: process.env.AWS_REGION });

/**
 * 获取指定DynamoDB表的监听字段列表
 * @param {string} tableName - 表名
 * @returns {Promise<string[]>} 监听字段数组
 */
async function getWatchedFields(tableName) {
  // 构造表的ARN（可通过DescribeTable API自动获取，这里直接构造更高效）
  const tableArn = `arn:aws:dynamodb:${process.env.AWS_REGION}:${process.env.AWS_ACCOUNT_ID}:table/${tableName}`;
  
  try {
    const command = new ListTagsOfResourceCommand({ ResourceArn: tableArn });
    const { Tags } = await dynamoClient.send(command);
    
    // 查找并解析field_names标签
    const fieldTag = Tags?.find(tag => tag.Key === "field_names");
    if (!fieldTag?.Value) return [];
    
    // 处理标签值，去重、过滤空字段
    return fieldTag.Value.split(",")
      .map(field => field.trim())
      .filter(field => field.length > 0);
  } catch (err) {
    console.error("获取标签失败:", err);
    return [];
  }
}

module.exports = { getWatchedFields };

3. Lambda中处理DynamoDB流并校验变更字段 #

在Lambda触发函数中，对比流记录的新旧数据，仅处理标签指定字段的变更：

const { unmarshall } = require("@aws-sdk/util-dynamodb");
const { getWatchedFields } = require("./tag-utils");

exports.handler = async (event) => {
  // 从流记录中提取表名
  const tableName = event.Records[0].eventSourceARN.split("/").pop();
  const watchedFields = await getWatchedFields(tableName);
  
  if (watchedFields.length === 0) {
    console.log("未配置监听字段，跳过处理");
    return;
  }

  for (const record of event.Records) {
    // 将DynamoDB格式的数据转换为普通JS对象
    const oldItem = record.dynamodb.OldImage ? unmarshall(record.dynamodb.OldImage) : {};
    const newItem = record.dynamodb.NewImage ? unmarshall(record.dynamodb.NewImage) : {};

    // 检查监听字段是否发生变更
    const changedFields = watchedFields.filter(field => {
      return JSON.stringify(oldItem[field]) !== JSON.stringify(newItem[field]);
    });

    if (changedFields.length > 0) {
      console.log(`检测到监听字段变更: ${changedFields.join(",")}`);
      // 这里添加你的业务逻辑，比如通知、数据同步等
      // await processChange(changedFields, oldItem, newItem);
    }
  }
};

4. 配置Lambda权限 #

确保Lambda执行角色拥有dynamodb:ListTagsOfResource权限，添加以下IAM策略：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "dynamodb:ListTagsOfResource",
      "Resource": "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/*"
    }
  ]
}

5. 动态更新监听字段 #

直接修改DynamoDB表的field_names标签即可调整监听字段，比如使用AWS CLI：

# 更新监听字段为name和profession
aws dynamodb tag-resource --resource-arn arn:aws:dynamodb:us-east-1:123456789012:table/People --tags Key=field_names,Value=name,profession

# 添加nationality字段到监听列表
aws dynamodb tag-resource --resource-arn arn:aws:dynamodb:us-east-1:123456789012:table/People --tags Key=field_names,Value=name,profession,nationality

注意事项 #

• 标签值需保持逗号分隔格式，代码已处理多余空格和空值
• Lambda每次触发都会重新获取标签，标签变更后下一次触发立即生效
• 若需减少API调用次数，可考虑将标签值缓存到Lambda环境变量或外部缓存服务，但需处理缓存失效逻辑

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