Azure Cosmos DB Spark 连接器：吞吐量控制

Spark 连接器支持使用 Apache Spark 与 Azure Cosmos DB 通信。 本文介绍如何使用服务器端吞吐量存储桶或客户端吞吐量控制来限制吞吐量消耗。 在受支持的情况下且可启用预览时，首选 吞吐量存储桶。 Azure Cosmos DB 在服务器上强制实施分桶限制，而不是依赖于客户端 RU 跟踪和协调。

本文中的客户端示例使用多区域吞吐量控制组来协调 Spark 客户端之间的请求单位（RU）消耗。 Java SDK 还支持全局组、单个客户端实例中的本地组和服务器端吞吐量存储桶。

为什么吞吐量控制很重要？

吞吐量控制有助于隔离针对容器运行的应用程序的性能需求。 吞吐量控制限制特定 Spark 客户端可以使用的 RU 量。

多个高级方案受益于吞吐量限制：

• 不同的操作和任务具有不同的优先级：由于数据引入或复制活动，可能需要防止正常事务受到限制。 某些操作或任务对延迟不敏感，与其他操作或任务相比，也更能容忍被限流。
• 为不同用户或租户提供公平性/隔离：一个应用通常有很多用户。 某些用户可能会发送过多请求，这会占用所有可用的吞吐量，并导致其他用户受到限制。
• 不同 Azure Cosmos DB 客户端之间的吞吐量负载均衡：在某些情况下，必须确保所有客户端获得合理（对等）的吞吐量份额。

吞吐量控制可根据需要实现更精细的 RU 速率限制。

首选服务器端吞吐量桶

吞吐量桶（预览版）限制工作负荷可以使用的容器吞吐量的最大百分比。 分配给同一桶的所有客户端的请求共享其限制。 当桶超出该限制时，Azure Cosmos DB 会对后续请求进行限制，并返回 HTTP 状态代码 429 和子状态代码 3212。

当你的帐户和容器支持时，对于 Spark 数据引入、迁移、批量写入、查询和变更馈送工作负载，优先使用 buckets。 服务器端强制实施可避免本文后面介绍的客户端近似和协调限制。 不需要单独的吞吐量控制元数据容器。

存储桶不保留吞吐量，也不保证其可以使用分配给它们的最大吞吐量。 吞吐量仍与其他工作负载共享。 每个容器最多提供 5 个存储桶，ID 为 1 到 5。

为 Spark 启用和配置存储桶

1. 将 Spark 连接器版本 4.43.0 或更高版本与符合您的 Spark 和 Scala 版本的构件一起使用。 请参阅 Spark 连接器设置说明。

2. 在 Azure 门户中订阅页上的“预览”功能下注册吞吐量存储桶。 等待审批和启用；启用可能会在审批后一小时内完成。

3. 在数据资源管理器中，打开容器的“缩放和设置”窗格，选择“吞吐量存储桶”，激活存储桶，并设置其最大吞吐量百分比。 在以下示例中，激活桶 2。 请参阅 配置吞吐量桶。

4. 将以下选项添加到现有的 Spark 读取或写入配置中，该配置包含帐户连接、数据库和容器选项：

   val throughputBucketOptions = Map(
       "spark.cosmos.throughputControl.enabled" -> "true",
       "spark.cosmos.throughputControl.name" -> "SparkBucket2",
       "spark.cosmos.throughputControl.throughputBucket" -> "2"
   )
   

   使用 .options(throughputBucketOptions) 将这些选项传递给 Spark 读取器或写入器。 连接器使用组名称将操作与存储桶相关联;它不会创建或重命名服务器端存储桶。 存储桶 ID 决定容器上配置的限制。

5. 当您从客户端吞吐量控制切换时，请删除 spark.cosmos.throughputControl.targetThroughput、spark.cosmos.throughputControl.targetThroughputThreshold 以及 spark.cosmos.throughputControl.globalControl.* 选项。 此外，如果您配置了单独的元数据帐户，请删除 spark.cosmos.throughputControl.accountEndpoint 和 spark.cosmos.throughputControl.accountKey。 不要将这些客户端设置与 throughputBucket 组合使用。 保留主帐户连接选项。

6. 在 Azure 门户中监视请求总数（预览版）和总请求单位（预览版）（按 ThroughputBucket 拆分），以验证工作负荷是否使用预期的桶。

使所选存储分区处于活动状态。 指定已删除的存储桶 ID 的请求使用容器的总体吞吐量，而不是受该存储桶的限制。 未分配存储桶的请求使用容器的默认存储桶（如果已配置）;否则，它们没有特定于存储桶的限制。

何时改用客户端吞吐量控制

当无法启用预览、需要绝对 RU/s 目标而不是百分比，或使用不受支持的帐户或容器时，请使用客户端吞吐量控制。 共享吞吐量数据库中的无服务器帐户或容器不支持吞吐量桶。 分配给存储桶的请求也不能使用突发容量。 有关详细信息，请参阅 吞吐量存储桶限制 和 常见问题。

客户端吞吐量控制的工作原理是什么？

网关模式下的客户端吞吐量控制需要最新版本的 Spark 连接器。 对于无服务器 Azure Cosmos DB 帐户，尝试使用targetThroughputThreshold定义百分比会导致失败。 相反，使用 设置绝对 RU/秒目标。

若要配置 Spark 连接器的吞吐量控制，请先创建一个定义吞吐量控制元数据的容器。 分区键为 groupId 并启用 ttl。 在这里，你将使用 Spark SQL 创建此容器，并将其命名为 ThroughputControl：

    %sql
    CREATE TABLE IF NOT EXISTS cosmosCatalog.`database-v4`.ThroughputControl 
    USING cosmos.oltp
    OPTIONS(spark.cosmos.database = 'database-v4')
    TBLPROPERTIES(partitionKeyPath = '/groupId', autoScaleMaxThroughput = '4000', indexingPolicy = 'AllProperties', defaultTtlInSeconds = '-1');

上面的示例创建具有自动缩放功能的容器。 如果你更喜欢标准预配，可以将 autoScaleMaxThroughput 替换为 manualThroughput。

分区键必须定义为 /groupId，并且必须启用 ttl，这样吞吐量控制功能才能正常工作。

在特定应用程序的 Spark 配置中，可以指定工作负载的参数。 以下示例将吞吐量控制设置为 enabled。 该示例定义吞吐量控制组 name 参数和 targetThroughputThreshold 参数。 您还需要定义用于维护吞吐量控制组的 database 和 container 参数：

    "spark.cosmos.throughputControl.enabled" -> "true",
    "spark.cosmos.throughputControl.name" -> "SourceContainerThroughputControl",
    "spark.cosmos.throughputControl.targetThroughputThreshold" -> "0.95", 
    "spark.cosmos.throughputControl.globalControl.database" -> "database-v4", 
    "spark.cosmos.throughputControl.globalControl.container" -> "ThroughputControl"

Microsoft Entra ID身份验证和吞吐量控制： 如果使用 spark 连接器Microsoft Entra ID（以前Azure Active Directory）身份验证，可能需要额外的基于角色的访问控制（RBAC）配置，具体取决于所使用的吞吐量控制参数：

• targetThroughputThreshold：必须向你的 Microsoft Entra 标识分配包含 Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/throughputSettings/read 操作的自定义角色。 连接器需要读取容器的预配吞吐量，以便根据阈值百分比计算目标 RU。
• targetThroughput：如果您使用早于 4.47.0 的 Spark 连接器版本，则还必须分配一个具有 Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/throughputSettings/read 操作的角色。 从版本 4.47.0 开始，使用 targetThroughput时不再需要此权限。

有关使用特定操作配置自定义角色的详细信息，请参阅为Azure Cosmos DB配置基于角色的访问控制。

在上一示例中，targetThroughputThreshold 参数定义为 0.95。 当客户端使用的分配给容器的吞吐量超过 95% (+/- 5-10%) 时，就会发生速率限制（并重试请求）。 此配置以文档的形式存储在吞吐量容器中，如以下示例所示：

    {
        "id": "ZGF0YWJhc2UtdjQvY3VzdG9tZXIvU291cmNlQ29udGFpbmVyVGhyb3VnaHB1dENvbnRyb2w.info",
        "groupId": "database-v4/customer/SourceContainerThroughputControl.config",
        "targetThroughput": "",
        "targetThroughputThreshold": "0.95",
        "isDefault": true,
        "_rid": "EHcYAPolTiABAAAAAAAAAA==",
        "_self": "dbs/EHcYAA==/colls/EHcYAPolTiA=/docs/EHcYAPolTiABAAAAAAAAAA==/",
        "_etag": "\"2101ea83-0000-1100-0000-627503dd0000\"",
        "_attachments": "attachments/",
        "_ts": 1651835869
    }

通过查询或更改源进行读取时，应将 spark.cosmos.read.maxItemCount 中的页面大小（默认为 1000）配置为适量。 这样，客户端吞吐量控制就可以使用更高的频率重新计算，并在任何特定时间更准确地反映。 当你对批量写入作业使用吞吐量控制时，单个请求中执行的文档数量将根据限制速率自动调整，以使吞吐量控制尽早开始。

对于使用吞吐量控制组的每个 Spark 客户端，将在 ThroughputControl 容器中创建一条记录，ttl 为几秒。 因此，如果 Spark 客户端不再主动运行，文档会很快消失。 下面是一个示例：

    {
        "id": "Zhjdieidjojdook3osk3okso3ksp3ospojsp92939j3299p3oj93pjp93jsps939pkp9ks39kp9339skp",
        "groupId": "database-v4/customer/SourceContainerThroughputControl.config",
        "_etag": "\"1782728-w98999w-ww9998w9-99990000\"",
        "ttl": 10,
        "initializeTime": "2022-06-26T02:24:40.054Z",
        "loadFactor": 0.97636377638898,
        "allocatedThroughput": 484.89444487847,
        "_rid": "EHcYAPolTiABAAAAAAAAAA==",
        "_self": "dbs/EHcYAA==/colls/EHcYAPolTiA=/docs/EHcYAPolTiABAAAAAAAAAA==/",
        "_etag": "\"2101ea83-0000-1100-0000-627503dd0000\"",
        "_attachments": "attachments/",
        "_ts": 1651835869
    }

在每个客户端记录中，loadFactor 属性表示特定客户端上相对于吞吐量控制组中其他客户端的负载。 allocatedThroughput 属性显示当前分配给此客户端的 RU 数量。 Spark 连接器将根据每个客户端的负载调整分配的吞吐量。 这样，每个客户端将获得与负载成比例的可用吞吐量份额。 客户端的目标是将其合计消耗控制在组分配额度内，但要遵循前面所述的客户端侧近似估算。

相关内容

• 了解吞吐量分桶
• 阅读吞吐量桶常见问题解答
• 浏览 GitHub 中的 Spark 示例
• 使用 Azure Cosmos DB Spark 连接器管理数据

• 详细了解 Apache Spark。