数据提取吞吐量最佳实践

本页介绍了将数据提取到 Cloud Healthcare API 时优化数据吞吐量的最佳实践。这些建议适用于有经验的技术从业者，他们有管理大型系统数据吞吐量的经验。

数据吞吐量

数据吞吐量是指 Cloud Healthcare API 每秒提取的资源（例如 FHIR 资源或 DICOM 实例）或字节数。

数据吞吐量限制

以下列表介绍了数据吞吐量可能受限的原因：

• 您未针对导致流量激增的大量请求进行规划。
• 带宽限制会减慢在短时间内发送的大量数据的提取速度。
• 多个并发事务会更改同一 Cloud Healthcare API 资源，从而导致数据争用。
• 发出的小型请求过多。如需了解详情，请参阅避免发出小型导入和导出请求。
• 并发运行的长时间运行操作 (LRO) 过多，并且带宽有限。
• 同时安排的 LRO 过多，导致失败。

重试失败的请求

如果客户端在请求失败后快速重试请求，则可能会超出 Cloud Healthcare API 配额。以下部分介绍了如何高效地重试失败的请求。

将指数退避算法与抖动和持久重试队列搭配使用

指数退避算法并引入抖动，这是一个针对网络应用的标准错误处理策略。客户端会定期重试失败的请求，并以指数级增加各次重试之间的延迟时间，同时添加一个随机的小延迟。

确保指数退避实现对于每次重试都是幂等的，尤其是在您使用自定义逻辑来绕过失败条件时。如需了解详情，请参阅 HTTP 规范中的 9.2.2 幂等方法。

大多数编程语言都提供了库，可简化指数级回退和类似重试策略的实现。对于长期或多进程重试，请实现持久重试队列。如果您超出最大退避时间，此队列可以重置重试机制。

重试以下请求时，请使用指数退避算法：

• 用于修改 FHIR 资源或 FHIR 资源软件包的操作。

• 同步 LRO 请求。如果 LRO 启动时出错或 LRO 失败，请重试。

  LRO 具有独特的错误，您可能需要实现以下重试策略：

  • 使用单独的软件包存储导入或创建操作失败的数据。
  • 针对未能处理的数据使用同步请求。

指数退避算法示例

指数退避算法以指数方式重试请求（不断增加各次重试之间的等待时间，直到达到最大退避时间）。以下算法使用抖动实现截断的指数退避算法：

1. 向 Cloud Healthcare API 发送请求。

2. 如果请求失败，请等待 1 + random-fraction 秒后再重试请求。

3. 如果请求失败，请等待 2 + random-fraction 秒后再重试请求。

4. 如果请求失败，请等待 4 + random-fraction 秒后再重试请求。

5. 继续此模式，每次重试后等待 2ⁿ + random-fraction 秒，最多 maximum-backoff 次。

6. deadline 秒后，停止重试请求。

在实现算法时，请使用以下值：

• 每次重试之前，等待时间为 min((2n + random-fraction), maximum-backoff)，其中 n 从 0 开始，并在每次重试时递增 1。

• 将 random-fraction 替换为小于或等于 1 的随机小数值。每次重试使用不同的值。添加此随机值可防止客户端同步和同时发送大量重试。

• 将 maximum-backoff 替换为重试之间等待的最长时间（以秒为单位）。典型值为 32 或 64（2⁵ 或 2⁶）秒。请选择最适合您的用例的值。

• 将 deadline 替换为持续发送重试的秒数上限。请选择一个反映您的用例的值。

客户端可以在达到 maximum-backoff 时间后使用与退避相同的值进行重试。例如，如果 maximum-backoff 时间为 64 秒，则每 64 秒重试一次。确保客户端不会无限重试。

使用流量整形实现客户端速率限制

速率限制可防止大规模系统因过多请求而超载，从而保护这些系统。如果客户端速率限制不够，Cloud Healthcare API 配额系统可能会限制数据吞吐量。如需了解详情，请参阅配额管理最佳实践。

如果您有其他要求（例如在重试后保证传送），重试失败的请求中的策略可能不够用。流量整形是一种速率限制技术，可将客户端请求的速率控制在带宽限制范围内。这样可以将负载高峰分布在数小时或数分钟内，从而提高吞吐量。当配额受限时，与仅使用重试相比，流量整形可以实现更高的吞吐量，因为它可以避免回推并跟踪工作器单元。

您可以针对同步创建、删除、更新和删除 (CRUD) 操作（包括 fhir.executeBundle）实现流量控制。

流量控制要求

如需实现流量整形，您的系统必须实现以下内容：

• 由存储空间支持且具有冗余性的处理队列，以避免磁盘故障。
• 协调工作器从处理队列中拉取数据。
• 总体使用情况检测功能可根据配额限制调整工作器数量及其处理速度。
• 存储空间支持的处理队列的灾难恢复。如果发生灾难，您的系统必须能够清除或恢复队列。
• 减少了高峰时段的 LRO。如需了解详情，请参阅高效规划和使用配额以及加入队列和管理 LRO。

在以下情况下，可能只需要对单个流水线阶段进行流量整形：

• 限制从上一个流水线步骤拉取数据的工作器数量。
• 单独限制每个工作器。
• 使用工作器池协调程序调整各个工作单元的处理速率，例如每秒查询次数 (QPS) 或每秒提取的字节数。

在系统的其他区域实现速率限制

您可以使用现有的编程语言和框架来实现流量整形。请考虑以下开源项目和预构建解决方案：

• Apache Beam 中的客户端节流。如需了解如何使用 numWorkers 和 maxNumWorkers 标志控制节流，请参阅横向自动扩缩。

• Google Guava 一组核心 Java 库中的 Java RateLimiter 类。

• Python ratelimiter 模块。

如需进行流控制，请使用高级 Pub/Sub 客户端库。

选择异步处理或同步处理

用于封装 Cloud Healthcare API 请求的客户端代理层（如在多个层级处理错误中所示）还可以控制使用 Cloud Healthcare API 的各项服务的节流。根据所需的流量整形类型，使用以下任一选项：

异步

使用异步处理将请求加入队列并控制工作器。代理层会将传入请求写入队列，并在每个请求加入队列后返回 200 OK 响应。这最适合写入请求，但如果客户端可以接收读取结果，则可以在 LRO 框架中用于读取请求。

同步

如果一项工作单元依赖于上一项工作单元的完成，同步处理可提供一种简单的反馈机制。代理层会根据 QPS 或字节吞吐量限制延迟出站请求，而客户端会阻塞并等待代理层的响应。

代理层可以根据实例数量调整速率限制，也可以与每隔几秒钟调整速率限制的控制器进程协调。为了让代理层跟踪实例数量及其速率限制，每个代理实例都可以定期读取文件或进行带有编码速率限制的远程过程调用 (RPC)。

同步处理有时会存在以下缺点：

• 在客户端阻塞并等待响应时，客户端和代理层中的资源不可用。这可能会导致错误、超时和数据吞吐量降低，从而使扩容变得更加困难。

• 如果客户端和代理层断开连接，则需要执行更多工作来确保数据已按要求修改。

使用 Cloud Tasks

使用 Cloud Tasks 将请求分流到队列。Cloud Tasks 会自动设置和监控以下 Google Cloud 配额：

• 使用 RateLimits 对象的突发大小上限和请求并发数上限
• 使用 RetryConfig 对象设置重试限制

如需在 Cloud Tasks 中创建队列，请参阅创建队列。Queue 资源会显示您可以在队列上设置的选项。例如，您可以使用 RetryConfig 对象实现指数退避算法。如需了解特定于语言的库，请参阅 Cloud Tasks 客户端库。

使用 Cloud Tasks 时，请考虑以下事项：

• Cloud Tasks 不保证“恰好传输一次”。在“恰好一次”传送模式下，服务器会将任何包含重复数据的请求识别为重复请求并忽略。如需了解详情，请参阅 Lambda 之后：Dataflow 中的“正好一次”处理，第 1 部分。
• 任务大小上限可能远小于 Cloud Healthcare API 中的最大 FHIR 软件包大小。如需了解详情，请参阅 Cloud Tasks 配额和限制以及 Cloud Healthcare API 配额和限制。
• Cloud Tasks 存在问题和限制。

将 FHIR 软件包与速率限制器结合使用

使用指数退避和速率限制器重试 FHIR 软件包有助于保持较高的数据吞吐量并管理负载高峰。

客户端可以将批量和事务 FHIR 软件包发送到 Cloud Tasks，后者会将软件包中的请求发送到 Cloud Healthcare API。如果速率限制器已满或超出配额（因为它已达到队列大小上限且用尽了磁盘空间），客户端可以实现指数退避算法来将软件包加入队列。

通过监控以下资源，防止速率限制器队列满载：

• Cloud Healthcare API 中的 FHIR 操作配额
• 速率限制器配额
• 速率限制器错误

如果速率限制器队列已满，您的系统必须向人工发出提醒，并阻止客户端发送请求。

使用 HTTP 持久（可重复使用的 keep-alive）连接

默认情况下，Cloud Healthcare API 会为每个 CRUD 请求打开新的 TCP 连接。这需要进行 TCP 握手，这可能会导致开销并降低性能。为了提高性能，请使用 HTTP keepalive 让 TCP 连接保持打开状态，以便处理多个请求。

如需在 HTTP/1.1 中使用 HTTP 保持连接，请将 Connection 标头设置为 keep-alive：

Connection: keep-alive

HTTP/2 使用一个 TCP 连接来处理顺序请求和并发请求，从而自动避免开销。

Python requests 库默认使用 HTTP 保持连接。如果您使用的是 Node.js，请在创建 http.Agent 对象时将 keepAlive 设置为 true，然后在请求中传递该对象。

使用测试框架

测试框架可确保您的代码正常运行，并帮助您执行以下操作：

• 为应用或流水线中的突然流量激增做好准备。
• 测试指数退避和客户端速率限制是否可以提高性能。测试可以显示这些实现是否会产生必须单独处理的任务积压。
• 分离和控制高优先级流量。例如，如果用户正在等待响应，则可以减少后台处理任务的工作负载，以确保用户体验不会受到影响。
• 测试同步和异步队列策略以调节流量，或测试代理层是否处理了推回。
• 规划灾难恢复。这通常需要重置传入流量，或在灾难结束后使用队列来恢复流量。

使用 Cloud Monitoring

使用 Cloud Monitoring 监控测试环境和生产环境。请遵循以下建议：

• 将 Cloud Tasks 与其他Google Cloud 日志记录和监控服务（例如 Cloud Audit Logs）集成。
• 使用 Cloud Monitoring API 创建自定义指标，以跟踪重试次数、队列大小和队列年龄等关键指标。
• 为您的环境创建服务等级目标 (SLO) 和服务等级指标 (SLI)。如需获取建议，请参阅 SLI 简介。
• 使用 Google Cloud Observability 创建提醒政策。提醒政策会在系统出现压力或需要人为干预等问题时通知您。
• 创建操作手册，以便系统管理员知道在提醒政策发送通知时该怎么做。

• 在预演环境中使用运营手册来应对以下场景：

  • 速率限制导致的积压
  • 因超出配额限制而导致的回推
  • 传入流量激增

防止 429 Resource Exhausted operation_too_costly 错误

每天对 FHIR 资源进行数千次并行更新可能会导致锁争用、延迟，并阻止事务完成。无法完成的交易可能会导致 429 Resource Exhausted operation_too_costly 错误积压：

HTTP/1.1 429 Too many requests
...

{
  "issue": [
    {
      "code": "too-costly",
      "details": {
        "text": "operation_too_costly"
      },
      "diagnostics": "aborted due to lock contention while executing transactional bundle. Resource type: FHIR_RESOURCE_TYPE",
      "severity": "error"
    }
  ],
  "resourceType": "OperationOutcome"
}

在该错误中，“费用”是指资源用量和数据吞吐量，而不是结算费用。

429 Too Many Requests 错误并不一定表示存在配额问题。当 Cloud Healthcare API FHIR 服务器检测到数据库记录存在过多锁争用时，可能会发生此错误。这可能是因为 FHIR 软件包中包含许多操作，或者是因为 CRUD 操作的组合。

请考虑以下场景：

1. 用于更新患者资源和其他 FHIR 资源的 FHIR 事务包会锁定患者资源，直到事务完成。

2. 多个 FHIR 软件包尝试并行更新 Patient 资源，并发生锁争用。错误响应包含文本为 Resource type: PATIENT 的 diagnostics 字段。

   您可以使用指数退避算法重试更新 Patient 资源，但长时间的锁争用期可能会导致超时、吞吐量降低和资源用量增加。

3. Cloud Healthcare API FHIR 服务器最终会检测到积压的事务，并通过返回 operation_too_costly 错误来进行负载分流。这会限制流量并防止进一步出错。

   operation_too_costly 错误会限制Google Cloud 项目中的所有 FHIR CRUD 操作，这会影响与您的项目关联的所有应用。

排查 429 Too Many Requests 错误

如需排查 429 Too Many Requests 错误，请搜索 Cloud Logging。包含 operation_too_costly 的错误表示锁争用。如果错误是由资源耗尽引起的，请检查配额问题。

如果发生节流，事务软件包可能会因锁争用程度较高而失败，并生成以下错误：

HTTP/1.1 429 Too many requests
...

{
  "issue": [
    {
      "code": "too-costly",
      "details": {
        "text": "operation_too_costly"
      },
      "diagnostics": "aborted due to cumulative heavy load or lock contention in this project while executing transactional bundle, please see https://cloud.google.com/healthcare-api/docs/troubleshooting#fhir_transaction_bundle_heavy_load for more information",
      "severity": "error"
    }
  ],
  "resourceType": "OperationOutcome"
}

如需排查此错误，请前往 diagnostics 字段中的 FHIR 事务包因累计高负载而中止链接。

避免使用较大的软件包

大型事务软件包更有可能出现 429 Too Many Requests 错误。任何大小的软件包都可能会造成吞吐量瓶颈。测试不同的软件包，以找到最佳大小。

包含重试的大型软件包可能会带来越来越少的性能回报，并且更容易出现多次失败。客户端应实现额外的逻辑来管理在事务中失败的 FHIR 资源子集。

如果批量软件包很大或 QPS 很高，则可能会遇到 429 Too Many Requests 和 413 Request Entity Too Large 错误以及吞吐量瓶颈。

避免使用包含数千笔交易的大型软件包。请改为执行以下操作：

• 使用支持数据一致性的较小事务软件包。如果 FHIR 资源彼此之间没有依赖关系，请单独更新这些资源。例如，FHIR 资源可能不依赖于同一软件包中的另一个资源的特定版本。
• 在软件包中使用一些批处理，并避免单独的请求。批处理可以提高性能，但大批处理可能会导致错误并降低数据吞吐量。大小相近的批处理软件包的争用较少，因为它们不会在 FHIR 资源更新期间保持锁定状态。

小型事务捆绑可以避免争用，因为它们一次只会持有几个锁，并且会快速完成。这有助于防止堆叠交易积压。

LRO 吞吐量

请参阅 LRO 数据吞吐量。

FHIR 数据存储选项

如果您的 FHIR 数据量较小到中等，请使用 fhir.create 存储数据。如需存储大量 FHIR 资源，请使用 fhir.executeBundle 或 fhirStores.import。如需了解每种方法，请参阅 FHIR 导入选项。

导入 FHIR 资源

在决定是否使用 FHIR 导入功能时，请考虑以下事项：

• FHIR 导入不会限制导入数据的总大小。如果 FHIR 软件包超过 50 MB，您可以将 FHIR 资源上传到 Cloud Storage 并进行导入。避免同时进行高延迟或大规模导入，否则数据吞吐量可能会受到限制。

• 与使用 FHIR 软件包相比，FHIR 导入的复杂性较低。例如，您无需执行以下操作：

  • 将大型软件包划分为较小的软件包
  • 管理时间表
  • 在资源或软件包级重试暂时性错误

• FHIR 导入不会强制执行参照完整性。如需了解详情，请参阅 FHIR 参照完整性。

• 如果数据新鲜度是重中之重，请勿使用 FHIR 导入功能。导入速度可能很快，但也可能会延迟数小时或数天。

• 当您的 Google Cloud 项目中包含的 LRO 较少时，FHIR 导入的效果会更好。

• 如果您的应用能够处理部分资源上的批量错误和失败，则 FHIR 导入可以实现高数据吞吐量。

使用 FHIR 软件包

在以下情况下，请使用 FHIR 软件包，而不是 FHIR 导入：

• 构建用于将数据存储在 Cloud Storage 中并导入数据的流水线的费用（结算费用或网络带宽费用）太高。

• 必须强制执行参照完整性。

• 必须强制执行 FHIR 配置文件验证。

• 存储 FHIR 资源时，您需要发送 Pub/Sub 通知。FHIR 导入不支持发布/订阅通知。

• 数据新鲜度是首要考虑因素，数据必须在几秒或几分钟内提取。 但是，即使在架构良好的系统中，数据吞吐量也可能会受到以下因素的限制：

  • 处理流水线的上游延迟。流水线可能需要更多时间来准备数据，然后才能提取数据。
  • 回退、重试和流量控制代理层。

FHIR 软件包存在以下限制：

• 配额和结算会应用于软件包中的每个操作，就好像每个操作都是单独执行的一样。例如，如果软件包包含 10 个 POST 操作、5 个 GET 操作和 1 个 DELETE 操作，则应用于该软件包的配额和结算方式与这些操作单独执行时相同。

• 大型事务包更有可能出现导致锁争用的事务冲突。如需了解详情，请参阅防止 429 Resource Exhausted operation_too_costly 错误。

• 批量软件包可以提高数据吞吐量，但不具备参照完整性等事务一致性功能。

• 大批量软件包可能会降低吞吐量。如需了解详情，请参阅避免创建较大的软件包。

DICOM 数据存储选项

在将数据从图片归档和通信系统 (PACS) 发送到 Cloud Healthcare API 时，您可以使用以下方法实现高数据吞吐量：

使用 DICOM 消息服务元素 (DIMSE) 协议的开源 Cloud Healthcare API DICOM 适配器

当您将 PACS 与 Cloud Healthcare API 同步时，适配器会优化数据吞吐量。在同步之前，请运行性能测试，并验证适配器是否可以持续提供峰值数据吞吐量。

如果您无法使用 Storage Transfer Service 或其他传输选项将 DICOM 文件上传到 Cloud Storage，请使用此适配器。例如，您可能无法满足以下 Storage Transfer Service 要求：

• 将文件系统装载到托管代理池中的代理的每台机器上，以检索来源数据。
• 如果您以定期间隔（而不是一次性批量加载）的方式传输数据，则必须衡量数据大小随时间的变化，以确定发生了哪些变化。
• 最大限度地提高代理转移性能。
• 支付和分配 Cloud Storage 存储空间。
• 验证数据是否已成功转移到 Cloud Storage。
• 将数据导入 Cloud Healthcare API 并修正所有导入错误后，移除 Cloud Storage 资源。
• 根据临床系统的网络和存储容量安排批量提取时间间隔。

我们建议您使用 Storage Transfer Service 进行单次批量加载，以填充 DICOM 存储区。定期使用 Storage Transfer Service 需要执行额外的工作，例如同步导入流水线。如需了解详情，请参阅 Storage Transfer Service 文件系统传输详情。

dicomStores.import

使用此方法存储大量 DICOM 数据。

DICOMweb 存储区事务

使用此方法可以编程方式存储 DICOM 数据。

管理配额以优化数据吞吐量

以下部分介绍了如何管理和规划配额以优化数据吞吐量。如需了解配额管理方面的常规最佳实践，请参阅配额管理最佳实践。

针对可预测的流量规划配额

首先分析客户端应用的典型每日流量，然后规划配额要求。即使流量是可预测的，也应规划比平均所需配额更多的配额。这有助于避免错误，并在流量激增或每日使用量偶尔增加时提供安全余量。

下图显示了大小一致且以可预测的模式发送的 Cloud Healthcare API 请求：

图 1. Google Cloud 项目

中所有数据集和数据存储空间的每小时 API 负载的汇总。

为大量请求规划配额

避免在高峰时段安排大型批量作业。如需了解详情，请参阅始终优先处理低交易量。

下图显示了一种可预测的流量模式。但是，在高峰流量期间发出大量批量请求会超出可用配额。这可能会导致项目中的所有请求都出现 429 Resource Exhausted 错误。

图 2. 高峰时段大型批量作业导致的资源使用量不规则分布。

如果您的系统有额外的灵活性配额，小流量激增不会导致错误，也不会导致可预测的峰值负载出现错误。这些小峰值必须分布在 Google Cloud 项目中产生负载的许多数据存储区、应用和其他客户端之间。

如需防止单个大型批量作业导致流量激增，请参阅避免创建大型软件包。

申请更多配额

如需保持较高的数据吞吐量并避免 429 Resource Exhausted 错误，请参阅本页中的最佳实践，尤其是管理配额以优化数据吞吐量。这些最佳实践可确保您的客户端应用坚如磐石，并可随着请求量变化而扩容。如果不实施最佳实践，仅申请更多配额，长期来看不太可能防止出现错误。

如果您已实施最佳实践，但仍需要更多配额，请参阅申请更多配额的最佳实践。

数据提取吞吐量资源

如需详细了解数据提取吞吐量，请参阅在 Google Cloud中管理工作负载的流量和负载。