DynamoDB 入门：用 Node.js 掌握 NoSQL 数据库 | 新手教程

在当今数据驱动的世界里，对可扩展、高性能且高可用的数据库的需求比以往任何时候都更强烈。Amazon DynamoDB 是 AWS（Amazon Web Services）提供的全托管、无服务器、宽列键值存储，旨在满足这些需求。无论您是在构建 Web 应用、移动应用，还是任何需要灵活且可靠存储的系统，DynamoDB 都能胜任。

本教程旨在带您认识 Amazon DynamoDB 的世界，并演示如何在 Node.js 应用中高效使用它。无论您是刚接触 NoSQL 数据库的初学者，还是希望扩展技能集的有经验开发者，本指南都会带您了解要点，并提供实用示例，帮助您快速上手。

什么是 DynamoDB？

Amazon DynamoDB 是一项无服务器的键值与文档数据库，提供无缝的可扩展性和低延迟性能。与传统关系型数据库不同，DynamoDB 无需您预置硬件或管理基础设施，非常适合需要根据负载变化动态扩缩的应用程序。

DynamoDB 的关键特性包括：

可扩展性

DynamoDB 每秒可处理数百万个请求，适用于流量水平变化的应用。

高可用性

通过跨多个可用区的内置复制，为高可用性和数据持久性而设计。

托管服务

AWS 负责运营开销，包括硬件预置、打补丁和备份，让您专注于构建应用程序。

灵活的数据模型

DynamoDB 同时支持键值和文档数据模型，为您的数据建模提供灵活性。

安全性

与 AWS 身份与访问管理（IAM）集成，可在条目和属性级别对数据库资源实施细粒度的安全访问控制。

成本节省

DynamoDB 的无服务器架构结合其自动扩缩能力，让您只需为实际消耗的资源付费，从而优化成本。无需前期硬件投入或长期承诺，对初创公司和企业均具成本效益。

数据类型

DynamoDB 支持的数据类型：

• 标量类型：包括字符串、数字、二进制、布尔和 null 值。
• 文档类型：支持带嵌套属性的复杂结构，非常适合处理 JSON 数据。
• 集合类型：字符串集合、数字集合和二进制集合，可高效存储相关值的集合。

为何将 DynamoDB 与 Node.js 搭配使用？

Node.js 是用于构建服务端应用（包括 Web 服务器和 API）的流行运行时环境。其非阻塞、事件驱动架构与 DynamoDB 的异步 API 非常契合，对 Node.js 开发者而言是天然之选。

在本教程中，我们将涵盖以下主题：

1. 在 DynamoDB 中创建表并定义模式。

2. 使用 AWS SDK for Node.js 执行基本 CRUD（创建、读取、更新、删除）操作。

3. 具备 ACID 保证的高级查询。

完成本教程后，您将对 Amazon DynamoDB 有扎实的理解，并能在 Node.js 应用中充分发挥其能力。现在，让我们开始探索这个多才多艺的 NoSQL 数据库吧！

准备工作

开始之前，您需要一个 AWS 账户。理论上无需 AWS 账户也可完成这些步骤，但为了更贴近真实场景，我们将使用 terraform 在 AWS 中创建基础设施。若要了解如何借助 Docker 在本地使用 AWS 服务，我提供了一个示例演示的 localstack GitHub 代码库，您可以参考并自行实验。

如果尚未安装 terraform，现在也请一并安装。

本教程余下部分假设您已使用您的凭证与配置文件完成 AWS CLI 的设置，并已安装 terraform。

上手

本教程将使用 TypeScript 和 yarn。我们先从简入手，使用 ts-node；在下一篇教程中我们将转向无服务器，并在此基础上使用 AWS Lambda 进行扩展。

要设置演示项目，新建一个文件夹，然后运行以下命令添加依赖并完成初始化：

yarn init
yarn add --dev typescript ts-node @types/node @types/ramda @types/uuid
yarn add @aws-sdk/client-dynamodb ramda uuid
npx tsc --init
mkdir src src/domain infra
touch ./src/index.ts ./src/client.ts ./infra/main.tf ./src/domain/student.ts
技术支持

接着将以下 scripts 段落添加到 package.json 中

  "scripts": {
    "dev": "ts-node src/index.ts"
  },
技术支持

main.tf

provider "aws" {
  region = "eu-west-1"
}


resource "aws_dynamodb_table" "demo-table" {
  name = "demo-table"
  billing_mode = "PAY_PER_REQUEST"
  hash_key = "pk"
  range_key = "sk"


  attribute {
    name = "pk"
    type = "S"
  }


  attribute {
    name = "sk"
    type = "S"
  }
}
技术支持

在命令行中 cd 到 infra 目录，运行 terraform init && terraform apply -auto-approve，然后前往 AWS 控制台查看您的 DynamoDB 表：

我会在另一篇文章中介绍 DynamoDB 的计费模式、WCU、RCU

client.ts

import { DynamoDB } from "@aws-sdk/client-dynamodb";


export const TABLE_NAME = "demo-table";
export const REGION = "eu-west-1";
export const dynamoClient = new DynamoDB({ region: REGION });


export const STUDENT_PREFIX = "student#";技术支持

数据模型

在继续之前，我们先定义一个本教程范围内要使用的数据模型。为便于理解，我选择建模为与 DataCamp 核心功能接近的结构。

本教程不会覆盖所有实体。我会在后续文章中扩展此模型。

使用 DynamoDB 进行 CRUD 操作

若要全面了解如何在 DynamoDB 中进行数据设计与建模，请参阅我的DynamoDB 单表设计教程。在本教程中，我将尽量简化对 DynamoDB 工作原理的解释。需要牢记的一点是，DynamoDB 使用分区键和排序键来高效管理数据。

分区键

分区键用于将数据分布到多个存储服务器。选择能均匀分布数据的分区键非常重要，以避免出现导致性能问题的“热点分区”（数据分布不均）。

常见且有助于均匀分布数据的分区键选择包括高基数字段，如：

• 唯一 ID
• 用户 ID
• 哈希值
• 地理区域
• 基于时间的键
• 复合键

分区键的选择在很大程度上取决于应用的访问模式和数据模型。

排序键

排序键是可选的，可用于在同一分区内组织数据。使用排序键可以在分区内高效查询与排序。

复合键

复合键同时由分区键和排序键组成。它们允许您创建更丰富的查询模式，例如按范围、时间或其他属性进行筛选。当数据访问模式需要复杂查询时，应使用复合键。

热点分区

为缓解热点分区问题，可考虑分片（为分区键使用随机前缀）、基于时间的分区，或均匀分布写入等策略。并通过监控与 AWS Auto Scaling 动态调整容量。

student.ts

我们将从简，使用学生 ID 同时作为分区键和排序键。这看起来可能有些奇怪，但我们需要保持模式灵活，以便支持多种数据关系与访问模式。我会在后续 DynamoDB 文章中扩展此模型。

import { head, omit, pathOr } from "ramda";
import { STUDENT_PREFIX, TABLE_NAME, dynamoClient as client } from "../client";
import { v4 as uuidv4 } from "uuid";
import {
  addPrefix,
  attributeMapToValues,
  attributeValueToValue,
  removePrefix,
  valueToAttributeValue,
} from "../utils";


const entityType = "student";


export const dynamoRecordToStudent = (record: any) => {
  const { pk, ...data } = record;


  return omit(["sk"], {
    ...attributeMapToValues(data),
    id: removePrefix(attributeValueToValue<string>(pk), STUDENT_PREFIX),
  });
};


export const getStudentById = (id: string) =>
  client
    .getItem({
      TableName: TABLE_NAME,
      Key: {
      pk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
      sk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
    },
  })
  .then(({ Item }) => (Item ? dynamoRecordToStudent(Item) : undefined));


export const saveStudent = async ({
  firstName,
  lastName,
  email,
}: {
  firstName: string;
  lastName: string;
  email: string;
}): Promise<string> => {
  const _id = uuidv4();
  const _email = email.toLocaleLowerCase();
  const xp = 0;


  await client.putItem({
    TableName: TABLE_NAME,
    Item: {
      pk: valueToAttributeValue(addPrefix(_id, STUDENT_PREFIX)),
      sk: valueToAttributeValue(addPrefix(_id, STUDENT_PREFIX)),
      firstName: valueToAttributeValue(firstName),
      lastName: valueToAttributeValue(lastName),
      email: valueToAttributeValue(_email),
      xp: valueToAttributeValue(xp),
      entityType: valueToAttributeValue(entityType),
    },
  });


  return _id;
};
技术支持

utils.ts

AWS 最新版的 DynamoDB SDK 提升了对存储在数据库中的条目的类型安全性，但也因此在使用上稍显繁琐。为缓解这一问题，我们来创建一些辅助函数。稍后当我们看到 DynamoDB 的内部存储方式时，就会明白它们的必要性。

在 src 目录根下新建名为 utils.ts 的文件。

import { AttributeValue } from "@aws-sdk/client-dynamodb";


export const valueToAttributeValue = <T>(value: T): AttributeValue => {
  switch (typeof value) {
    case "string":
      return { S: value };
    case "number":
      return { N: `${value}` };
    case "boolean":
      return { BOOL: value };
    case "object":
      if (Array.isArray(value)) {
        return { L: value.map((item) => valueToAttributeValue(item)) };
      }
      return {
        M: Object.entries(value as any).reduce(
          (acc, [key, item]) => ({
            ...acc,
            [key]: valueToAttributeValue(item),
          }),
          {}
        ),
      };
    default:
      throw new Error(`Unknown type ${typeof value}`);
  }
};


export const attributeValueToValue = <T>(value: AttributeValue): T => {
  switch (true) {
    case !!value.S:
      return value.S as T;
    case !!value.N:
      return Number(value.N) as T;
    case !!value.BOOL:
      return value.BOOL as T;
    case !!value.L:
      return value.L?.map((item) =>
        attributeValueToValue(item)
      ) as unknown as T;
    case !!value.M:
      return Object.entries(value.M || []).reduce(
        (acc, [key, item]) => ({ ...acc, [key]: attributeValueToValue(item) }),
        {}
      ) as unknown as T;
    default:
      throw new Error(`Unknown type ${JSON.stringify(value)}`);
  }
};


export const attributeMapToValues = (
  items: Record<string, AttributeValue>
): unknown[] =>
  Object.keys(items).reduce(
    (acc, key) => ({
      ...acc,
      [key]: attributeValueToValue(items[key]),
    }),
    []
  );


export const removePrefix = (id: string, prefix: string): string =>
  id.replace(prefix, "");


export const addPrefix = (id: string, prefix: string): string =>
  `${prefix}${removePrefix(id, prefix)}`;
技术支持

创建新学生

终于可以开始向 DynamoDB 发送 CRUD 操作的 API 调用了！

打开 index.ts 文件，粘贴以下代码，然后在命令行运行 yarn dev。

import { getStudentById, saveStudent } from "./domain/student";


Promise.resolve()
  .then(async () => {
    const id = await saveStudent({
      firstName: "John",
      lastName: "Smith",
      email: "[email protected]",
    });


    const john = await getStudentById(id);


    console.log(john);
  })
  .catch((err) => {
    console.error(err);
    process.exit(1);
  })
  .then(() => {
    console.log("done");
    process.exit(0);
  });
技术支持

您应会看到类似如下的返回结果：

{
  "entityType": "student",
  "id": "dd4c2ee4-9422-4957-ae6f-f9fff748e5ab",
  "firstName": "John",
  "lastName": "Smith",
  "email": "[email protected]",
  "xp": 0
}
技术支持

现在使用 CLI 扫描 DynamoDB，运行：

aws dynamodb scan --table-name demo-table --no-cli-pager

从输出中您可以看到 DynamoDB 内部如何存储和标注数据类型，其中 N 表示数字类型，S 表示字符串类型，这也能解答您对于 utils.ts 中代码的一些疑问。

{
  "Items": [
    {
      "entityType": {
        "S": "student"
      },
      "lastName": {
        "S": "Smith"
      },
      "email": {
        "S": "[email protected]"
      },
      "xp": {
        "N": "0"
      },
      "sk": {
        "S": "student#dd4c2ee4-9422-4957-ae6f-f9fff748e5ab"
      },
      "pk": {
        "S": "student#dd4c2ee4-9422-4957-ae6f-f9fff748e5ab"
      },
      "firstName": {
        "S": "John"
      }
    }
  ],
  "Count": 1,
  "ScannedCount": 1,
  "ConsumedCapacity": null
}
技术支持

通过电子邮件地址获取学生

如果我们只有学生的电子邮件地址，如何找到对应记录？我们可以通过添加全局二级索引来支持这种访问模式。

首先，我们需要销毁原有表（为了避免今后每次 terraform 变更都这么做，我们稍后会引入一个清空表的函数）：

terraform destroy

接着，更新 terraform，在 demo-table 上添加以下 GSI：

attribute {
  name = "gsi1_pk"
  type = "S"
}


attribute {
  name = "gsi1_sk"
  type = "S"
}


global_secondary_index {
  name = "gsi1"
  hash_key = "gsi1_pk"
  range_key = "gsi1_sk"
  projection_type = "ALL"
}
技术支持

通过以下命令应用变更：

terraform apply -auto-approve

接下来，我们需要对保存学生的方式做一个小改动：

export const saveStudent = async ({
  firstName,
  lastName,
  email,
}: {
  firstName: string;
  lastName: string;
  email: string;
}): Promise<string> => {
  const _id = uuidv4();
  const _email = email.toLocaleLowerCase();
  const xp = 0;


  await client.putItem({
    TableName: TABLE_NAME,
    Item: {
      pk: valueToAttributeValue(addPrefix(_id, STUDENT_PREFIX)),
      sk: valueToAttributeValue(addPrefix(_id, STUDENT_PREFIX)),
      gsi1_pk: valueToAttributeValue(_email),
      gsi1_sk: valueToAttributeValue(addPrefix(_id, STUDENT_PREFIX)),
      firstName: valueToAttributeValue(firstName),
      lastName: valueToAttributeValue(lastName),
      xp: valueToAttributeValue(xp),
      entityType: valueToAttributeValue(entityType),
    },
  });


  return _id;
};
技术支持

注意我们不再显式存储 email，而是将其用作 GSI 的分区键值。

我们需要对 DynamoDB 学生数据的转换方式做一个小调整：

export const dynamoRecordToStudent = (record: any) => {
  const { pk, gsi1_pk, ...data } = record;


  return omit(["sk", "gsi1_sk"], {
    ...attributeMapToValues(data),
    id: removePrefix(attributeValueToValue<string>(pk), STUDENT_PREFIX),
    email: attributeValueToValue<string>(gsi1_pk),
  });
};
技术支持

最后，我们可以使用新建的 GSI gsi1 通过电子邮件地址查询学生记录：

export const getStudentByEmail = (email: string) =>
  client
    .query({
      TableName: TABLE_NAME,
      IndexName: "gsi1",
      KeyConditionExpression: "#gsi1_pk = :gsi1_pk",
      ExpressionAttributeNames: {
        "#gsi1_pk": "gsi1_pk",
      },
      ExpressionAttributeValues: {
        ":gsi1_pk": {
          S: email.toLocaleLowerCase(),
        },
      },
    })
    .then((res) => head(pathOr([], ["Items"], res).map(dynamoRecordToStudent)));技术支持

更新学生

要更新学生，需要使用 update 方法，并同时提供分区键和排序键，以唯一标识要更新的条目。

export const updateStudent = async ({
  id,
  firstName,
  lastName,
  email,
}: {
  id: string;
  firstName?: string;
  lastName?: string;
  email?: string;
}) => {
  const updateExpressionParts = [];
  const ExpressionAttributeValues: Record<string, any> = {};


  if (firstName !== undefined) {
    updateExpressionParts.push("#firstName = :firstName");
    ExpressionAttributeValues[":firstName"] = valueToAttributeValue(firstName);
  }


  if (lastName !== undefined) {
    updateExpressionParts.push("#lastName = :lastName");
    ExpressionAttributeValues[":lastName"] = valueToAttributeValue(lastName);
  }


  if (email !== undefined) {
    updateExpressionParts.push("#gsi1_pk = :gsi1_pk");
    ExpressionAttributeValues[":gsi1_pk"] = valueToAttributeValue(email);
  }


  const UpdateExpression = `SET ${updateExpressionParts.join(", ")}`;


  const ExpressionAttributeNames = {
    ...(firstName && { "#firstName": "firstName" }),
    ...(lastName && { "#lastName": "lastName" }),
    ...(email && { "#gsi1_pk": "gsi1_pk" }),
  };


  await client.updateItem({
    TableName: TABLE_NAME,
    Key: {
      pk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
      sk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
    },
    UpdateExpression,
    ExpressionAttributeNames,
    ExpressionAttributeValues,
  });
};
技术支持

删除数据

首先需要说明，DynamoDB 并不原生支持截断表，除非删除整个表再重建（就像我们之前为引入新 GSI 所做的那样）。我会在 client.ts 中添加一个清空方法，但要强调这仅用于测试目的，不能在生产环境使用。原因是为了知道要删除哪些数据，我们必须扫描整张表；而在 DynamoDB 中每个操作都会产生成本。如果您的表中有上百万条记录，那么直接删除表并重新开始会更便宜（当然需要配合某种数据迁移流程，以免丢失生产数据）。以下是我们的清空函数：

type AttributeMap = Record<string, AttributeValue>;


const getItemKeyAndValue = (item: AttributeMap, key?: string) =>
  key ? { [`${key}`]: item[`${key}`] } : {};


export const truncateTable = async (
  client: DynamoDB,
  TableName: string,
  hash: string,
  range?: string
): Promise<void> => {
  const { Items } = await client.scan({ TableName });
  if (!Items) {
    return;
  }
  const keys = Items.map((item: AttributeMap) => ({
    ...getItemKeyAndValue(item, hash),
    ...getItemKeyAndValue(item, range),
  }));
  if (!keys.length) {
    return;
  }
  await Promise.all(keys?.map((Key) => client.deleteItem({ TableName, Key })));
};
技术支持

清空 demo-table：

await truncateTable(dynamoClient, TABLE_NAME, "pk", "sk");

若要删除某个特定学生，我们实际上不希望直接使用 SDK 的 deleteItem 方法。NoSQL 数据库没有参照完整性的概念，如果数据库中还有其他条目引用了某个已不存在的学生，这将成为问题；而 DynamoDB 会在我们指示时毫无阻拦地删除条目，不会像关系型数据库那样因为存在外键引用而在删除操作时报错。

为稳妥起见，我们将实现软删除，并对两个读取函数做些小修改。

export const deleteStudent = async (id: string) => {
  await client.updateItem({
    TableName: TABLE_NAME,
    Key: {
      pk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
      sk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
    },
    UpdateExpression: "SET #deleted = :deleted",
    ExpressionAttributeNames: {
      "#deleted": "deleted",
    },
    ExpressionAttributeValues: {
      ":deleted": valueToAttributeValue(true),
    },
  });
};
技术支持

getStudentByEmail 的查询现在应当过滤已删除的记录。需要注意的是，这种“过滤”发生在客户端而非数据库端，这意味着该查询在检索已删除记录时仍会产生相同的成本。

{
  TableName: TABLE_NAME,
  IndexName: "gsi1",
  KeyConditionExpression: "#gsi1_pk = :gsi1_pk",
  ExpressionAttributeNames: {
    "#gsi1_pk": "gsi1_pk",
  },
  ExpressionAttributeValues: {
    ":gsi1_pk": {
      S: email.toLocaleLowerCase(),
    },
    ":notDeleted": {
      BOOL: false,
    },
  },
  FilterExpression:
    "attribute_not_exists(deleted) OR deleted = :notDeleted",
 }
技术支持

getStudentById 使用的是 getItem，因此无法应用过滤表达式，我们将显式写出逻辑：

export const getStudentById = (id: string): Promise<Student | null> =>
  client
    .getItem({
      TableName: TABLE_NAME,
      Key: {
        pk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
        sk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
      },
    })
    .then(({ Item }) => {
      if (!Item) {
        return null;
      }
      const _item = dynamoRecordToStudent(Item);
      return _item.deleted ? null : _item;
    });
技术支持

具备 ACID 保证的原子更新

DataCamp 为每位学生收集 XP 数据，以提供激励、成就感、整体进度跟踪的便捷方式，以及衡量与同侪差距的手段。在关系型数据库中，我们有多种工具来处理与聚合数据，但在 NoSQL 数据库中并非如此。尽管某些 NoSQL 数据库（如 MongoDB）提供聚合框架，但在数据库规模巨大时，这些工具往往会力不从心；而选择 NoSQL 的原因往往正是为了支持海量数据与流量。

在 DynamoDB 中，原子更新指能够对某个条目的单个属性执行特定更新操作，并确保满足原子性、一致性、隔离性和持久性（ACID 属性）。原子更新可以在存在并发更新的情况下，仍然保证对特定属性值的修改操作成功。

利用这种技术，当用户完成学习目标时，我们可以轻松为其增加 XP。

export const updateStudentXp = async ({
  id,
  xp,
}: {
  id: string;
  xp: number;
}) => {
  await client.updateItem({
    TableName: TABLE_NAME,
    Key: {
      pk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
      sk: valueToAttributeValue(addPrefix(id, STUDENT_PREFIX)),
    },
    UpdateExpression: "set xp = xp + :inc",
    ExpressionAttributeValues: {
      ":inc": valueToAttributeValue(xp),
    },
  });
};
技术支持

如果这段代码看起来眼熟，那是因为我们在 deleteStudent 函数中已经用过这种模式！

结语

本教程介绍了由 AWS 提供的无服务器、高度可扩展数据库 Amazon DynamoDB 的基础知识。其无缝扩展性、高可用性与灵活的数据建模，使其成为现代应用的优秀选择。

我们讲解了创建表、通过 Node.js 执行 CRUD 操作等重要主题，并讨论了为保持数据一致性而进行原子更新的重要性。在整个教程中，我们构建了数据模型并实现了与 DynamoDB 交互的函数。

在继续学习 DynamoDB 的过程中，建议深入了解单表设计、性能优化以及用于实时数据处理的 DynamoDB Streams 等高级主题。DynamoDB 的多样性与可扩展性，将助力您在 AWS 生态中构建高性能、无服务器的应用程序。

您可以在Github 上找到本教程的完整源代码。祝编码顺利！

延伸学习资料

• DataCamp 教程：DynamoDB 单表设计
• DataCamp 课程：NoSQL 概念
• AWS re:Invent：DynamoDB 高级设计模式
• Rick Houlihan：DynamoDB 单表设计基础
• Alex DeBrie：DynamoDB 之书