Bulk Executor for Amazon DynamoDB, Part 2: Lessons we learned the hard way

Amazon DynamoDB 批量执行器，第 2 部分：我们从艰难教训中学到的经验 | AWS 架构博客

Amazon DynamoDB 批量执行器，第 2 部分：我们从艰难教训中学到的经验

在《介绍开源 Amazon DynamoDB 批量执行器》博客文章中，您了解了批量执行器工具的目的和用法。在本文中，我们将分享构建该工具的历程：设计权衡、意外发现、陷阱以及最终的解决方法。可以将其视为您使用 Amazon DynamoDB 和 AWS Glue 的旅行指南。

设计要求

批量执行器是一个旨在对表中许多或所有项目执行批量操作的项目。例如，您可以批量删除旧项目或比较两个表的相等性。

三个核心需求塑造了后续的大部分决策：运行批量命令应感觉像本地脚本（实时输出，可通过 ^C 中断），它应能扩展到数百个并行工作线程而无需用户管理基础设施，并且应能轻松添加新命令而无需与框架紧密耦合。

选择 AWS Glue

我们选择 AWS Glue 来驱动并行执行。这是一个无服务器服务，您提交作业后，这些作业会分发到多个工作线程（计算实例），每个工作线程并行执行任务。有关更多信息，请参阅 Apache Spark 的关键主题。我们的命令行客户端启动一个作业，AWS Glue 负责将工作分发到可能多达数百个工作线程。

我们还考虑了 Amazon Simple Queue Service（Amazon SQS）配合 AWS Lambda、AWS Fargate、Amazon EMR 和 Ray。我们决定使用 AWS Glue，因为它为我们提供了无服务器并行性和内置的 DynamoDB 连接器，因此我们无需付出太多成本就获得了许多复杂功能。以下简短的代码示例展示了如何读取或写入项目：

# 从 DynamoDB 读取到 Spark 数据框
dataframe = (
    spark.read
    .format("dynamodb")
    .option("dynamodb.input.tableName", "source")
    .option("dynamodb.throughput.read", "40000")
    .option("dynamodb.consistentRead", "false")
    .load()
)

# 从 Spark 数据框写入 DynamoDB
(
    dataframe.write
    .format("dynamodb")
    .option("dynamodb.output.tableName", "sink")
    .option("dynamodb.throughput.write", "40000")
    .option("dynamodb.item.size.check.enabled", "true")
    .save()
)

Spark 数据框为我们提供了所需的内置功能。具体包括：

• 一个 where/filter，接受 Spark SQL 条件来过滤数据。
• 一个 orderBy，用于对数据进行排序。
• 一个 limit，用于控制项目数量。

基于这些功能，我们构建了 find 命令作为我们的第一个命令，其 --where、--orderby 和 --limit 参数几乎无需我们编写代码。如果您想打印最近的 100 个未送达订单，可以发出如下批量命令：

./bulk find --table orders --where "status = 'undelivered'" --orderby timestamp desc --limit 100

count 命令使用相同的代码，但会统计项目数量而不是打印它们。delete 命令也是如此，但找到项目后会并行执行批量删除调用以删除它们。

Spark 连接器的一个缺点是它不能完全表示源表的数据类型。它简化了数据，例如，不会区分字符串集合和字符串列表。

在处理大型表时，我们希望在开始工作前提供成本预估。预测未来总是充满挑战。以下是我们的实现方式。

对于读取操作，我们查看表的元数据（大约每6小时更新一次，因此请注意刚加载的表可能会显示比实际更小的大小），以了解其近期大小并计算完整扫描所需的读取请求单元数。对于按需表，计算方式很简单：调用定价API获取该区域每百万读取请求单元的成本，然后相乘即可。

对于预置模式，成本取决于预置级别、自动扩展余量和任务持续时间，这使得估算更加困难。我们的启发式方法是取按需价格的三分之二（除以1.5）。原理是：预置在100%利用率的表成本约为等效按需吞吐量的30%。但实践中很难达到100%利用率。在50%利用率时，成本约为60%。我们采用67%的估算值，因为略微高估比低估更安全。

对于删除命令，写入成本取决于匹配项的数量，这需要扫描后才能确定。因此删除命令分两个阶段计算价格：首先提前估算读取成本，然后在找到匹配项后，根据删除集的大小计算写入成本。

对于填充命令（根据自定义生成器批量加载合成数据），我们首先生成一些示例项以了解大致大小，然后据此定价。

我们不估算AWS Glue数据处理单元（DPU）成本有两个原因。首先，无法简单地确定需要多少计算资源。其次，我们观察到对于大型任务，DPU成本通常只占DynamoDB成本的一小部分。这很合理，因为以较低每小时成本运行的AWS Glue工作者可以驱动大量数据库操作。对于小型任务，DPU成本占比更高，但绝对金额较低。你可以指定--XWaitForDPU参数，在执行后查看实际DPU成本报告。

每个--X参数都是针对批量执行器本身的参数，而不是正在运行的命令。我们参考了Java虚拟机的约定进行设计。

作业终止(^C)

我们需要一种方法让调用者能够终止处理，例如当成本预估结果高于预期时。按下^C会立即停止客户端Python脚本，但AWS Glue作业会持续运行直到超时。

我们通过让Python代码程序化捕获中断信号来解决这个问题。中断处理程序检测到^C后会设置一个中断标志。这个标志会被始终运行的监视循环检测到。通常它会监视AWS Glue状态变化等事件。但当检测到中断时，它会主动指示AWS Glue作业停止，然后再进行客户端程序的干净退出。现在一个^C可以同时干净地终止两边。

强制启用时间点恢复(PITR)

像删除这样的命令有可能删除超出预期的数据。我们设计选择要求对任何接收变更命令的表必须启用时间点恢复(PITR)。这样，如果您的--where子句没有完全达到预期效果，您就有办法回滚。

这是我们在构建基于 PITR 驱动的增量导出的撤销导出（revert-export）命令时考虑的一个原因，该命令可以基于增量导出结果直接在原地撤销写入操作。请参阅我们之前的博客文章“使用批量执行器从意外的 DynamoDB 更改中恢复”。

监控执行（CloudWatch 实时日志）

在我们的第一个版本中，客户端会轮询 Amazon CloudWatch Logs 以获取 AWS Glue 工作器的输出。这种方式效率低下，导致输出出现延迟并以块状形式呈现。切换到 Amazon CloudWatch Live Tail 后，作业输出可以近乎实时地流式传输到客户端。

这引入了一个问题：当 find 命令只返回少量匹配项时，一切正常；但面对大量结果时，输出会不完整。根本原因在于：Live Tail 每秒最多显示 500 个事件，超过这个数量后，它只会流式传输部分匹配事件。

我们的解决方案是将 find 或 diff 命令的大规模输出写入 Amazon Simple Storage Service（Amazon S3）存储桶，仅将前 10 个结果流式传输到终端，并附上指向完整数据集的链接。

Live Tail 还带来了两个额外的挑战。事件可能无序到达，因此我们构建了一个重新排序层。此外，AWS Glue 作业可能在最终日志事件出现在 CloudWatch 中之前就完成（存在几秒的传播延迟），因此我们不能在作业停止的瞬间立即停止监听。我们实现了一个“微波炉爆米花监控器”：只有在几秒静默后才停止监听。

快速且实用的错误处理

当 AWS Glue 任务发生错误时，它会记录异常，默认情况下堆栈跟踪会进入日志供客户端查看。如果错误仅出现在单个任务中，这尚可管理，但略显冗长。但如果错误是所有任务都遇到的问题（如错误的 --where 子句），您将看到数百条相同的堆栈跟踪信息淹没终端。我们通过四种模式解决了这个问题。

在客户端，每个命令在启动 AWS Glue 作业前都会验证参数。拼写错误会本地捕获，而无需 AWS Glue 启动任何任务。

在 AWS Glue 内部，我们创建了一个 error_accumulator，这是一个 Spark 累加器。我们不是让异常从任务中大规模传播，而是让每个任务将异常添加到累加器中。Spark 驱动程序随后提取第一个错误并仅打印一次：一条干净的提示信息，而不是数百条重复信息，随后进行优雅的作业关闭。

在客户端持续监控方面，Live Tail 的事件流会经过一个模式匹配器，寻找永远不会自行解决的错误模式（如 AccessDeniedException、ModuleNotFoundError、OutOfMemoryError），并可立即触发作业停止，而不是让每个任务都以相同错误失败。

此外，为了防止像 --where 子句中拼写错误这样简单的问题出现“可怕的”堆栈跟踪，我们使用了一种特殊的特定于应用的异常类型，当命令希望提示错误而非堆栈跟踪时可以选择抛出。只有开发者类型的错误才应生成堆栈跟踪。

经过一段时间的探索，我们开发出了一种我们喜欢的错误处理方式：尽可能快速地捕获错误，将其简化为仅向用户显示必要的信息，并在每次执行的最后一条信息中明确指示作业是否成功或失败（原因是什么）。我们还添加了颜色编码（红色表示错误，黄色表示警告），使问题更容易被发现。

DynamoDB 连接器与直接访问的对比

AWS Glue 连接器可以读写 DataFrame，但没有删除数据项的功能。对于删除操作，我们别无选择，只能直接调用 DynamoDB API。该命令逻辑首先使用连接器查找匹配项（生成一个 DataFrame），然后遍历这些项并通过批量写入执行删除操作。以下是简化版本：

def delete_partition(partition, table_name, keys):
    table = dynamodb_resource.Table(table_name)
    with table.batch_writer() as batch:
        for record in partition:
            item = json.loads(record)
            key = {k: item[k] for k in keys}
            batch.delete_item(Key=key)

# 跨工作节点分布删除操作
records.toJSON().foreachPartition(
    lambda partition: delete_partition(partition, table_name, keys)
)

每个 Spark 分区会获取匹配项的一个切片，foreachPartition() 会在所有工作节点上并行执行删除函数。batch_writer() 在底层会将单个删除操作封装成高效的批量写入调用。

我们构建 fill 和 load-export 命令时，也选择直接使用 batch_writer 写入数据，而不是构建 DataFrame。

这种直接调用的模式效果如此显著，促使我们重新思考最初命令的设计方式。我们开发了 scancount 命令来探索是否可以在不使用连接器的情况下实现更快的计数操作。

答案是：始终更快，因为减少了开销。基于连接器的计数需要将每个项目加载到 AWS Glue 上下文中，然后进行过滤和计数。而 scancount 使用 DynamoDB 分段扫描配合下推谓词和 Select: COUNT 参数，数据库只返回已应用过滤条件后的计数结果本身。每个工作节点并行扫描表的不同分段：

def _count(table_name, index_name, filter_expression,
             segment, total_segments, total_matched_accumulator, error_accumulator):
    table = dynamodb_resource.Table(table_name)

    scan_kwargs = {
        "Select": "COUNT",
        "Segment": segment,
        "TotalSegments": total_segments
    }
    if index_name:
        scan_kwargs["IndexName"] = index_name
    if filter_expression:
        scan_kwargs["FilterExpression"] = filter_expression

    # 省略错误处理代码
    local_count = 0
    while True:
        response = table.scan(**scan_kwargs)
        local_count += response.get("Count", 0)
        if "LastEvaluatedKey" not in response:
            break
        scan_kwargs["ExclusiveStartKey"] = response["LastEvaluatedKey"]

    total_matched_accumulator.add(local_count)
    return local_count

parallelize_count = 100
rdd = spark_context.parallelize(range(parallelize_count), parallelize_count)
rdd.foreach(lambda worker_id: _count(table_name, index_name,
    filter_expression, worker_id, parallelize_count,
    total_matched_accumulator, error_accumulator))

权衡取舍：scancount 接受原始的 --filter-expression 参数，而不是 Spark SQL 的高级 --where 语法，因为不涉及 DataFrame。

diff 和 copy 命令的代码也采用直接读取方式，而非使用连接器。开发这些命令时并非为了速度，而是为了确保数据类型保真度。

关于 diff 的补充说明：每个任务必须直接将其输出子集写入自己的 Amazon S3 对象，而不能将输出集中回驱动节点，否则无法扩展。

速率限制

连接器提供了内置的速率限制功能。然而，对于直接调用 API 的命令（如 delete、fill、copy 和 scancount），并没有速率限制。如果足够多的 AWS Glue 工作节点在没有约束的情况下运行循环，通常会产生不必要的节流。

我们构建了一个跨所有工作节点协调的分布式速率限制器。详细方法请参见《使用 Python Boto3 对 Amazon DynamoDB 进行速率限制，第 1 部分和第 2 部分：分布式协调》。

用户通过 --XMaxReadRate 和 --XMaxWriteRate 参数控制该功能。如果未指定参数，系统会根据账户和表的设置及配额推断合适的速率。

最近我们新增了一个功能：如果最大读取或写入速率无法在 --XTimeout（默认为 1 小时）内完成任务，将发出黄色警告。提前警告用户更有价值：如果即使最大速率也无法在超时时间内完成任务，用户可以重新启动任务并设置更高的速率或更长的超时时间。

构建命令库

我们最初的设想是让命令像新 Unix 命令一样易于使用。这种松耦合的设计带来了原始核心团队之外的贡献。到目前为止，我们已收到以下贡献：

• diff 命令由 Chris Gillespie 贡献。
• sql 命令由 John Terhune 贡献。
• load 命令由 Arjan Schaaf 贡献。
• load-export 和 revert-export 命令由 Ruskin Dantra 贡献。

每个命令都是一个独立的 Python 模块，通过名称加载。接口设计简洁：导出 run(job, spark_context, glue_context, parsed_args) 函数，并遵循共享约定（使用 error_accumulator 处理异常，使用分布式速率限制器处理 DynamoDB 调用，以及使用共享库辅助成本估算）。

在 AI 协助下升级到新版本 AWS Glue

几个月前，我们从 AWS Glue 4 升级到 AWS Glue 5。优势包括：作业启动速度更快，以及新的 DynamoDB 连接器可以直接加载到 DataFrame（旧连接器需要通过 DynamicFrame 中间步骤，增加了开销和复杂性）。

这次升级也让我们有机会将新旧连接器进行对比测试。我们使用 AI 工具生成了一套压力测试用例，在 AWS Glue 5 环境下测试不同表大小和读写模式的表现。AI 运行了数小时测试，收集结果并生成对比报告，最终确认升级是值得的。

这次升级原本在待办事项列表中搁置了许久，我们一直推迟执行。借助 AI 自动生成和执行测试，我们仅用一次交互式工作会话就完成了升级。

此后我们又升级到 AWS Glue 6，这次是为了利用 DPU 小时成本降低 30% 的优势。

实现 AI 代码检查器

我们发现有些内容很难通过单元测试或端到端测试进行验证。例如，我们有部分数据以三种不同格式存储在三个不同位置，需要确保这三个定义不会出现偏差。

首先，我们有一个 README 文件解释了安全相关的一些内容。您可以让系统为您生成一个新的 AWS Glue 作业角色，或者通过名称传入自定义角色。该文件以英文描述了如果自行创建 AWS Glue 作业角色，所需的最低 AWS Identity and Access Management（IAM）权限。这是其中一个定义。

第二个定义是，当我们创建角色时，会将正确的权限集附加到该角色上。第三个定义是，当你提供角色时，我们会检查该角色的有效权限（部分通过IAM策略模拟器，部分通过读取其附加的策略），以确认你已为其赋予了适当的最小权限。这遵循了这样一个原则：如果未满足条件，我们应在此处尽早报错，而不是在首次执行时才报错。

如果随着权限需求的变化，三个不同表述方式的相同核心事实出现分歧，这将是一个隐蔽且严重的错误。将英文描述与创建角色的逻辑和IAM检查逻辑进行对比并不容易。

我们的解决方案是创建一个AI代码检查工具。它是一个项目文件夹，其中包含一个定义检查工具行为的README文件，以及一个子文件夹，该子文件夹中包含一组以Markdown文件形式保存的规则。当需要时，我们可以要求AI运行代码检查，无需任何编码操作。

我们还有其他规则，例如每条命令（包括未来可能出现的命令）都应实施速率限制，无论通过连接器还是Python库。这虽然难以测试，但AI可以轻松推理。另一条规则是，所有输出量较大的命令必须将输出写入Amazon S3（遵循特定的位置选择约定），并向终端输出一个带有链接的有限预览。

任何误报信息对我们开发团队来说仅是参考信息，我们可以完善规则以消除这些误报。当然，AI还负责编写这些Markdown文件本身。

结论

我们解决的每个问题本身都不是巨大的难题。它们都是些小问题：Live Tail采样事件、^C无法终止远程任务、错误信息重复数百次、无法提前预知成本、无限制的读写速率等。但每个问题在修复前都会让工具的使用变得令人沮丧。如果你正在基于AWS Glue构建自己的工具，希望这些解决方案能为你节省一些探索时间。我们也希望我们使用AI进行长期或难以执行的测试的经验，能为你带来一些启发。

该项目在GitHub上是开源的，我们欢迎贡献。现在你已经了解了内部实现，你已经具备编写自己命令所需的一切。编写run()函数，并依赖框架处理其余部分。如果你认为这个项目具有普遍实用性，请给我们提交一个pull request！

作者简介

'"`