How to Turn Long-Running Python Simulations Into a Self-Service Platform

如何将长时间运行的 Python 模拟转化为自助服务平台 | Towards Data Science

数据工程

如何将长时间运行的 Python 模拟转化为自助服务平台

在 AWS 上构建异步工作负载过程中，我学到的关于队列、故障恢复和可重复性的经验

Anand Rawat

2026年10月21日

17分钟阅读时间

在 Google 上关注我们

分享

图片由作者提供

只要只有一个人运行，长时间运行的 Python 模拟看起来就是一个计算问题。程序运行在能够处理它的机器上，完成时会生成输出。如果运行速度变慢或中途失败，通常的解决方案是使用更大的机器或仔细查看堆栈跟踪，当你独自面对时，虽然麻烦但尚可处理。

一旦其他人开始依赖它，这种安排就会崩溃。多个人希望提交模拟任务，任务耗时数小时，而请求到达的速度却快于它们完成的速度。有些运行会在中途失败，迟早会有人要求获取上个月的结果并期望得到相同的答案。人们还想知道他们的任务是处于等待、运行、失败还是完成状态，没有人希望一个任务干扰另一个任务，也不希望在队列为空时机器仍在消耗资源。这些与提升程序速度关系不大。

因此问题发生了变化。到那时，运行 Python 程序只是简单部分，更困难的是将其转化为其他人无需考虑底层基础设施即可使用的东西。

我相信真正的工程挑战在于程序能运行与人们可信赖的服务之间的差距。这不仅限于模拟，模型评估和批量处理也会遇到相同的问题，大多数长时间运行的 Python 工作也是如此。运行一次代码很简单。为许多人重复运行代码，使每次运行都能被找到、追踪和重复，同时又不干扰其他运行，这是一项完全不同的工作。

我在模拟基础设施中亲身经历了这一点，本文其余部分将从内部视角描述这一过程。

瓶颈并非计算能力

在担任亚马逊英国软件开发工程师期间，我参与了基于 AWS 的交通模拟基础设施开发，旨在使实验更加可扩展、可重复且成本效益更高。回顾那段工作，最突出的一点是计算机本身并不是难题所在。

我们最初采用的是几乎完全手动的工作流程，操作人员起着关键作用。输入、配置和基础设施设置都是手工完成的，工程师亲自运行模拟，每当运行失败时，他们花费更多时间关注输出和“哪里出错了”的问题。计算资源的分配方式相同，无论是否运行了有用的负载都会进行分配。

当对模拟的需求增长时，压力立即显现。每个新请求在消耗计算资源之前，都需要消耗某个人的注意力。运行缓慢或失败的模拟难以追踪，因为证据分散在输入、配置和运行它的机器中。在某个时刻我意识到，限制因素变成了工作所需的协调量，而机器数量与之关系甚微。

更多机器只会提高理论上限，却无法解决排队、运行之间的隔离或重复执行等问题。可靠状态、故障后的恢复、可追溯性和可重复性仍将无法解决，而空闲容量只会持续增长。我提出的问题在过程中发生了变化。我停止思考如何让这个运行得更快，转而思考如何让多个人安全地提交工作、查看每个任务的执行状态，并在出现问题时进行恢复。

本文其余部分的论点是：一旦有超过一个人依赖于长期运行的模拟，真正的问题就变成了作业控制，这实际上是一系列更小问题的集合：排队、状态、隔离、故障恢复、可重复性以及一个不依赖任何单个任务的计算生命周期。AWS服务是我们实现这一目标的方式，我将在后文多次提及它们，但它们是最不值得关注的部分。

关于这些技术的来源，有一点需要说明。此处的专业经验来自我在亚马逊英国的工作经历。代码和实验属于一个小型本地参考实现，它是为了教育目的而独立创建的，灵感来自于那项工作的普遍经验。它并未复制亚马逊的生产架构，也并不代表亚马逊的观点。

断开提交与运行之间的联系

我记忆最深的决定是将作业提交与模拟执行分离。一个面向用户的过程启动长期模拟并等待其完成，这会拥有过多的控制权。虽然没有东西阻止它验证输入，但提交、验证、执行、基础设施生命周期和作业状态最终都会被焊接在一起，因为单个请求掌控了整个长期运行的过程。

这个请求变成了运行的枷锁。连接中断或重启可能会连带使工作失败，计算存在的时间问题与用户按下提交按钮的时间问题被混为一谈。将提交与执行分离，为每个作业提供了清晰的边界。

因此，我们首先存储了配置和输入引用，为每个运行分配了唯一的作业ID，并将任务放入SQS。基于Lambda的编排系统会提取队列中的任务，并为模拟启动一个独立的EC2执行环境。S3保存了持久化构件（输入、配置、日志和输出），因此大型对象不再依赖于机器保持运行，而Redshift则保存了人们希望跨运行比较的结构化结果。Lambda适合处理短小的控制平面任务，而模拟需要更多对运行时、内存和依赖项的控制，因此它们运行在EC2上。

人们常说队列可以扩展，这解释了很少的内容，所以这里是我们从队列中获得的实际效果。它将需求与容量相遇的点进行了转移。提交的爆发变成了一个积压队列，以容量允许的任何速度逐渐处理，而不是一次性全部落在计算层上。这就是反向压力，用户无需考虑它。异步边界还为我们提供了更清晰的位置来引入验证、状态跟踪、重试、死信处理和监控。回顾过去，这次改变让我们从一堆脚本和机器中，走向了可复用的解决方案。

它还引发了本文大部分内容所讨论的问题。标准的 SQS 队列至少会传递一次每条消息，这意味着相同的消息可能会出现两次，所有下游处理都必须以此为前提进行设计。

在本地版本中，我使用 SQLite 同时作为队列和持久化状态的存储。这是一个权宜之计，现在坦白承认这一点比日后被指出要好。这样设计的目的是让你无需 AWS 账户即可运行所有内容。

队列表代表异步传递，jobs 表保存持久化任务状态。dead_letter 表扮演死信队列（DLQ）的角色，子进程下的每个尝试目录作为执行边界。events 表记录生命周期历史。工作负载是一个刻意保持简单的蒙特卡洛模拟，使用固定种子确保相同配置生成相同结果。所有数据均为模拟生成。模拟根据配置和种子自行生成随机传递时间，因此不涉及外部数据集或亚马逊数据。

提交流程是我希望让顺序显而易见的地方。以下是从演示代码的 submit 方法中提取的片段，经过轻微格式调整，省略了参数检查和数据库插入部分：

python

with self.connect() as conn:
    if idempotency_key:
        existing = conn.execute(
            "SELECT job_id FROM jobs WHERE idempotency_key = ?",
            (idempotency_key,),
        ).fetchone()
        if existing:
            self.record_event(conn, existing["job_id"], "IDEMPOTENT_SUBMISSION_REUSED",
                              {"idempotency_key": idempotency_key})
            return existing["job_id"]
    job_id = f"sim_{uuid.uuid4().hex[:12]}"
    job_dir = self.jobs_root / job_id
    job_dir.mkdir(parents=True, exist_ok=False)
    config_path = job_dir / "config.json"
    config_path.write_text(json.dumps(asdict(config), indent=2) + "\n")
    # (INSERT INTO jobs ... status 'SUBMITTED' trimmed here)
    self.record_event(conn, job_id, "SUBMITTED", {"config_hash": config_hash(config)})
    conn.execute(
        "UPDATE jobs SET status='QUEUED', updated_at=? WHERE job_id=?",
        (utc_now(), job_id),
    )
    self._enqueue(conn, job_id)
    self.record_event(conn, job_id, "QUEUED")
    return job_id

配置信息在任何消息存在之前就已写入磁盘，消息中仅包含任务 ID。因此工作节点永远不会被分配到无效指针，队列也不会携带数据集。

不过我对顶部的查询逻辑有些疑虑。插入前的查询只能防止连续重试。相同键的两个请求同时到达时，都可能绕过该检查，这就是为什么持久化唯一性约束比应用层检查更可靠。演示代码中 idempotency_key 列声明为 UNIQUE 正是出于这个原因，但代码未捕获约束冲突，我也没有测试相同键的并发提交。因此这是一个普遍性考虑，而非我实际观察到的问题。生产系统需要有意处理这种竞争条件，例如通过捕获约束冲突并返回已存在的任务。

关于 SQLite 的替代方案再补充一点。在 SQLite 中，任务行和队列消息是在同一个事务中写入的。SQS 和独立的状态存储无法共享同一事务，因此在 AWS 上，这两个写入的顺序需要认真考虑。

状态管理比我预期的更复杂

在我们构建的所有功能中，确保作业状态在异步组件之间保持可信比预期的要困难得多。反复出现的问题是：系统是否能够信任它对作业状态的判断。

仅因为收到消息、EC2实例启动、Python进程退出或出现输出文件，模拟就不会结束。要判定作业完成，平台必须知道请求已被验证且执行确实开始，当前运行的是哪次尝试，是否已达到检查点。它还必须知道输出是否已写入、分析结果是否可用，以及作业是否已进入有效的终端状态。

基础设施状态与作业状态是两回事，当它们被混为一谈时问题就出现了。一个正在运行的实例并不能说明模拟真的在运行，返回零的进程也不能证明作业应被标记为完成。部分生成的产物更不具说服力，因为运行过程可能留下大量文件却从未完成。

故障处理是最复杂的部分。部分执行后消息可能被重新投递，计算过程可能在运行中途消失，粗心的重试可能重复工作或让未完成的结果看起来像成功。我们通过唯一作业标识符、显式状态转换和幂等性，配合持久化状态和执行记录、检查点机制、有限重试和死信处理来应对这些问题。重试安全性尤其依赖于能跨尝试保留的状态，以及只能发生一次的启动状态转换。

在演示中，这个启动状态转换出乎意料地简单。队列消息只是提示可能有工作存在，而作业行才是最终决定者。实际的process_one方法包含更多分支，因此此处简化为关键部分：

# 从process_one简化
conn.execute("BEGIN IMMEDIATE")  # 读取状态前获取写锁
job = conn.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
# 一旦作业处于终端状态或已运行，重复投递是无害的
if job["status"] in TERMINAL_STATES or job["status"] == "RUNNING":
...  # 确认消息，记录DUPLICATE_MESSAGE_IGNORED，提交，返回
attempt = int(job["attempts"]) + 1
conn.execute(
"""
UPDATE jobs
SET status='RUNNING', attempts=?, updated_at=?
WHERE job_id=? AND status='QUEUED'
""",
(attempt, utc_now(), job_id),
)
self.record_event(conn, job_id, "RUNNING", {"attempt": attempt})
conn.commit()

为了验证，我提交了相同的幂等性键两次，却得到了原始作业ID而非第二个作业。作业完成后，我注入了重复的队列消息，系统确认并忽略了它。记录的生命周期显示：SUBMITTED → QUEUED → IDEMPOTENT_SUBMISSION_REUSED → RUNNING → RESULT_PERSISTED → COMPLETED → 最终DUPLICATE_MESSAGE_IGNORED。

接下来是重试机制，只有当系统知道前一次尝试留下的状态时，重试才是安全的。每次尝试都有独立目录，因此失败尝试仍可保留用于检查，而下一次尝试可以干净启动。完整的失败处理方法包含更多簿记逻辑，但关键决策逻辑如下：

# 从 _handle_failure 简化而来
error = f"{type(exc).__name__}: {exc}"
self.record_event(conn, job_id, "ATTEMPT_FAILED", {"attempt": attempt, "error": error})
conn.execute("UPDATE queue SET state='ACKED' WHERE message_id=?", (message_id,))
if attempt < max_attempts:
...  # 将任务重新设为 QUEUED 状态并保留错误信息
# 为本地参考实现添加短暂的确定性延迟
self._enqueue(conn, job_id, delay_seconds=0.01)
self.record_event(conn, job_id, "REQUEUED", {"next_attempt": attempt + 1})
return
... # 将任务设为 FAILED 状态，并向死信表插入记录
self.record_event(conn, job_id, "DEAD_LETTERED", {"error": error})

在短暂性实验中，首次尝试故意失败。任务被重新入队，并在第二次尝试中成功，完整生命周期包括 SUBMITTED、QUEUED、RUNNING、ATTEMPT_FAILED、REQUEUED、RUNNING、RESULT_PERSISTED 和 COMPLETED。

在永久性实验中，所有尝试均失败。预算为2次时，任务经历 RUNNING、ATTEMPT_FAILED、REQUEUED、RUNNING、ATTEMPT_FAILED，随后进入 DEAD_LETTERED 状态，并最终标记为 FAILED，同时在死信表中生成记录。第二次运行对我而言更为重要，因为无限重试本质上只是缓慢失败的方式。下图展示了这三个实验所涉及的生命周期。

本地演示中的任务生命周期。图片由作者提供。

确定性失败（如无效配置或不兼容输入）通过再次尝试不会改善。演示在此方面非常粗略。它会持续重试所有失败直到尝试次数耗尽，而无效配置仅在任务提交阶段（任务尚未存在时）被检测到。真实系统应首先对失败进行分类，仅对值得重试的失败类型消耗尝试次数。

演示存在实际限制。它仅处理仍在运行的工作者内部抛出的异常，这大约是它的处理边界。它未模拟工作者在运行中途消失的情况，也没有检查点机制（这是我们处理计算中途终止的方式之一）或执行超时（生产环境工作者应强制实施该机制以防止进程挂起无限占用资源）。

最大的缺陷是本地队列没有可见性超时，也没有回收被遗弃工作的机制。如果工作者在获取消息后消失，消息不会自动返回，任务可能卡在 RUNNING 状态，其他工作者也没有自动机制来接管该任务。

仅因任务显示 RUNNING 就忽略重传消息也可能很危险，因为任务可能已不再 RUNNING，而缺失的回收机制才是导致任务卡住的根源。在 SQS 中，可见性超时会在超时后将未确认的消息重新放回队列，死信队列则收集持续失败的消息。这描述了服务当前的工作方式。我并未声称这与我们最初系统的设置方式一致。

演示还规定当进程干净退出且存在结果文件时标记任务为 COMPLETED。这是一个很低的标准。平台不应仅因工作者完成执行或调用返回就报告成功。它应确认预期输出已生成、正确存储，并与正确的输入和执行版本相关联。

隔离性使并行处理变得安全

增加更多工作者本身并不能保证并发安全。独立任务也需要避免相互干扰，而它们失败的方式往往非常普通。任务可能会共享临时文件或争夺依赖项，有时一个任务会继承前一次运行的残留内容，或者在崩溃时拖累邻居。在我们的实现中，每个模拟都在自己的 EC2 执行环境中运行。这为它提供了独立的配置、执行上下文、临时状态和资源边界，以及独立的日志和输出路径。如果某次运行耗尽了资源或失败，也不会影响其他运行，且失败原因可以追溯到该任务和实例。我不会一般性地认为按任务分配 EC2 是最佳选择。但它确实为任务之间划出了一条明确界限，而这一点我将在后文再次提及。

本地演示为每次尝试使用一个子进程，每个子进程都在自己的目录中运行。为了测试并发性，我提交了四个使用不同种子的独立任务。一个工作者依次完成它们耗时 14.623 秒，而四个工作者仅耗时 4.652 秒，速度提升了 3.14 倍。这个数字在被截图前需要一个说明。它来自一次本地参考实验，结果取决于环境和核心数量。这不是 AWS 的基准测试，与本文后文的生产结果也没有任何关联。

我更关心的是所有四个任务都完成了，每个任务都保留了自己的任务目录，不同的种子产生了不同的结果校验和。独立任务能够并发运行，而不会互相覆盖或污染彼此的任务特定状态和输出。

尽管如此，子进程作为边界远不如独立机器强。这四个任务仍然共享同一个主机、同一组 CPU 和内存资源，以及同一个 SQLite 数据库，因此该演示仅展示了任务范围内的状态，对资源隔离问题未作任何说明。

## 保存的结果不等于可复现的运行

如果我能回到过去并更强调某一点，那将是对这一点的强调。保存的结果只能告诉你结果是什么，而可复现的实验可以告诉你哪些数据、参数、代码和环境产生了这个结果。如果今天重新构建这个系统，不可变的运行清单和血缘关系将从一开始就作为核心特性存在，而我不会描述原始平台具备这些特性。

演示通过多层结构构建了这一点。配置哈希标识了请求的内容，而结果校验和标识了返回的内容。校验和刻意排除了墙钟运行时间，这样即使两个等效运行在不同速度的机器上执行，仍然可以进行比较。如果运行时间是校验和的一部分，那么任何两个运行都不会匹配：

def stable_result_checksum(result):

运行时间是操作元数据，不属于科学结果。

stable = {k: v for k, v in result.items() if k != "runtime_seconds"} payload = json.dumps(stable, sort_keys=True).encode() return hashlib.sha256(payload).hexdigest()[:16]

然后清单记录了执行本身，包括模拟代码的哈希值，因为没有代码的校验和只能带你走到一半：

manifest = { "job_id": job["job_id"], "attempt": attempt, "config_hash": result["config_hash"], "result_checksum": result["result_checksum"], "config_file_sha256": sha256_file(Path(job["config_path"])), "result_file_sha256": sha256_file(result_path), "simulation_code_sha256": sha256_file(simulation_script), "python_version": platform.python_version(), "platform": platform.platform(), "execution_seconds": round(elapsed, 4), "completed_at": utc_now(), }

为了验证这个机制，我提交了两个具有不同作业ID但配置和种子完全相同的任务。两个任务都成功完成，并在配置哈希和结果校验和（b7d726357add275f）上达成一致，各自生成了独立的执行清单。这揭示了两个容易混淆的概念：两次执行无需共享操作身份即可在科学上完全一致。它们的历史可以完全不同，但结果仍然等价。

这里的限制同样值得关注。该演示没有输入数据集，因此从未测试过版本化输入。清单记录了Python版本和平台信息，但未包含模拟依赖的库信息。更全面的生产清单应包含依赖版本、锁文件数据，或更理想的容器镜像摘要。我也不应假设不同数值库版本或平台之间的校验和完全一致，这就是为什么清单需要明确指定环境。

## 也要适度缩减计算规模

原始平台仅在有任务时才启动计算资源。拥有持续运行的模拟能力可以带来更低的启动延迟和更简单的调度，但当没有模拟运行时需要支付空闲资源的费用。按需启动则能削减这部分空闲成本，但需要接受启动延迟，并增加更多编排、更强的健康检查、超时控制，以及最重要的可信赖的作业状态管理。在我们的设计中，EC2实例会在任务到达时启动，任务完成后立即终止。

回顾来看，成本效率和可靠性最终相互强化。任务完成后消失的实例不能保留任何有价值的数据。基于需求的计算因此强制了更清晰的作业边界和持久化状态，这反过来意味着独立执行和显式的生命周期管理。

我们的项目报告和基准测试记录中有两个结果与此相关。通过比较相似的工作周期，团队使用可复用框架运行的模拟实验数量大约是早期主要依赖人工操作的工作流的三倍。单纯增加机器数量并未带来这种提升。真正起作用的是自助提交、异步排队、隔离执行以及更少的人工协调。空闲模拟计算成本降至接近零，但仅适用于EC2执行层，因为S3、SQS、监控和Redshift仍然保持其常规成本。

这两个数字与上面的本地实验无关。是的，3.14倍和大约3倍看起来相似，但这只是巧合，本地运行无法验证生产环境的数值。另外，通过系统化基准测试实现的可重复模拟场景，使机器学习交付预测性能相比既定基准提升了18%，这一成果应归功于建模工作，而非队列或实例本身。

## 现在我会做出的不同选择

我仍然会坚持这些原则。用于存储工件的持久化对象存储、带有反压机制的异步缓冲、按需计算，以及控制平面与执行平面的清晰分离，这些设计依然合理，尤其是最后这一点。

在执行层，我会优先评估AWS Batch或ECS，而不是直接管理EC2环境。容器化的作业定义能够明确指定依赖项、资源分配、可重复性和清理工作，而不是依赖脚本来处理这些内容。对于计算密集型且依赖队列的工作负载，Batch会表现得非常好。它自带作业队列和重试策略，因此本文中的一些自定义代码可以迁移到服务中，但作业身份和证据记录仍需要我们自行设计。我还会尝试使用Step Functions来管理验证、资源配置、执行、检查点恢复、后处理和故障升级的整个生命周期，因为重试和状态转移可以在一个地方统一可见，而不是分散在自定义的编排代码中。

我不会仅仅因为Kubernetes功能强大就选择它。它的灵活性确实存在，但运营成本和控制平面的复杂性同样显著，因此只有在组织已经使用Kubernetes并有更广泛的使用需求时，我才会考虑采用。同时，我会从第一天起就将运行声明、血缘关系和可重复性作为平台的核心组成部分。

···

## 最后思考

回顾整个项目，计算能力从来都不是难点。模拟可以在EC2工作节点上并行运行，也可以在AWS Batch管理的计算资源或本地子进程中执行。使平台值得信赖的关键在于将每个模拟视为一个受管理的工作单元，具有明确的身份、生命周期、独立的执行环境、有限的故障行为，以及对其发生过程和到达状态的完整记录。这种思维方式与我最初思考如何运行模拟时的视角完全不同。最终，重点变成了其他人如何能够依赖这个平台。对于另一位工程师来说，简而言之，多用户模拟平台更接近作业控制和证据系统，而不是一组机器的集合。

增加计算资源可以让模拟并行运行，但唯有明确的作业状态、隔离性、可重复性和故障恢复机制，才能将它们转化为可靠的自助服务平台。