使用 Azure 流分析中的查询并行化

本文说明了如何利用 Azure 流分析中的并行化。 了解如何通过配置输入分区和调整分析查询定义来缩放流分析作业。

作为先决条件，建议先熟悉了解并调整流单元中所述的流单元的概念。

流分析作业定义包括至少一个流式处理输入、查询和输出。 输入是作业读取数据流的地方。 查询是用于转换数据输入流的一种方式，而输出则是作业将作业结果发送到的地方。

利用分区，可根据分区键将数据分为多个子集。 如果输入（例如事件中心）是根据某个键进行分区的，强烈建议在向流分析作业添加输入时指定此分区键。 缩放流分析作业时，可利用输入和输出中的分区。 流分析作业可以并行使用和写入不同的分区，从而增加吞吐量。

所有 Azure 流分析的流输入都可以利用分区：事件中心、IoT 中心、Blob 存储和 Data Lake Storage Gen2。

处理流分析时，可利用输出中的分区：

• Azure Data Lake Storage
• Azure Functions
• Azure 表
• Blob 存储（可显式设置分区键）
• Azure Cosmos DB（需显式设置分区键）
• 事件中心（需显式设置分区键）
• IoT 中心（需显式设置分区键）
• 服务总线
• 使用可选分区的 SQL 和 Azure Synapse Analytics：请在“输出到 Azure SQL 数据库”页中查看详细信息。

Power BI 不支持分区。 但是，你仍可如本节中所述，对输入进行分区。

若要深入了解分区，请参阅以下文章：

• 事件中心功能概述
• Data partitioning（数据分区）

若要使某个作业并行，分区键需要在所有输入、所有查询逻辑步骤和所有输出之间保持一致。 查询逻辑分区由用于连接和聚合 (GROUP BY) 的键确定。 如果该查询逻辑并非键控查询逻辑（投影、筛选器、引用连接等），则可以忽略最后一个要求。

• 如果输入和输出按 WarehouseId 进行分区，并且查询按 ProductId 分组而不使用 WarehouseId，则作业不是并行的。
• 如果要连接的两个输入按不同的分区键（WarehouseId 和 ProductId）进行分区，则作业不是并行的。
• 如果单个作业中包含两个或多个独立数据流，每个数据流都有自己的分区键，则作业不是并行的。

仅当所有输入、输出和查询步骤都使用相同的键时，作业才会并行。

易并行作业是 Azure 流分析中最具可伸缩性的方案。 它将查询的一个实例的输入的一个分区连接到输出的一个分区。 实现此并行需满足以下要求：

• 如果查询逻辑取决于同一个查询实例正在处理的相同键，则必须确保事件转到输入的同一个分区。 对于事件中心或 IoT 中心，这意味着事件数据必须已设置 PartitionKey 值。 或者，可以使用已分区的发件人。 对于 Blob 存储，这意味着事件将发送到相同的分区文件夹。 例如，按 userID 聚合数据的查询实例使用 userID 作为分区键来对输入事件中心进行分区。 但是，如果查询逻辑不需要由同一个查询实例处理同一个键，则可忽略此要求。 举例来说，简单的选择项目筛选器查询就体现了此逻辑。

• 下一步是对查询进行分区。 对于兼容性级别为 1.2 或更高级别（推荐）的作业，可以在输入设置中将自定义列指定为分区键，作业将自动并行化。 兼容性级别为 1.0 或 1.1 的作业要求在所有查询步骤中使用“PARTITION BY PartitionId”。 允许采用多个步骤，但必须使用相同的键对其进行分区。

• 流分析中支持的大部分输出可利用分区。 如果使用不支持分区的输出类型，则作业不会易并行。 对于事件中心输出，请确保将“分区键列”设置为查询中使用的同一分区键。 有关详细信息，请参阅输出部分。

• 输入分区数必须等于输出分区数。 Blob 存储输出可支持分区，并继承上游查询的分区方案。 指定 Blob 存储的分区键时，数据将按每个输入分区进行分区，因此结果仍然是并行的。 下面是允许完全并行作业的分区值示例：

  • 8 个事件中心输入分区和 8 个事件中心输出分区
  • 8 个事件中心输入分区和 Blob 存储输出
  • 8 个事件中心输入分区和 Blob 存储输出（按具有任意基数的自定义字段进行分区）
  • 8 个 Blob 存储输入分区和 Blob 存储输出
  • 8 个 Blob 存储输入分区和 8 个事件中心输出分区

以下部分介绍一些易并行的示例方案。

• 输入：具有 8 个分区的事件中心
• 输出：具有 8 个分区的事件中心（“分区键列”必须设置为使用“PartitionId”）

查询：

    --Using compatibility level 1.2 or above
    SELECT TollBoothId
    FROM Input1
    WHERE TollBoothId > 100
    
    --Using compatibility level 1.0 or 1.1
    SELECT TollBoothId
    FROM Input1 PARTITION BY PartitionId
    WHERE TollBoothId > 100

此查询是一个简单的筛选器。 因此，无需担心对发送到事件中心的输入进行分区。 请注意，使用版本低于 1.2 的兼容性级别的作业必须包含 PARTITION BY PartitionId 子句才能满足上述要求 #2。 对于输出，需要在作业中配置事件中心输出，将分区键设置为“PartitionId”。 最后一项检查是确保输入分区数等于输出分区数。

• 输入：具有 8 个分区的事件中心
• 输出：Blob 存储

查询：

    --Using compatibility level 1.2 or above
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId
    
    --Using compatibility level 1.0 or 1.1
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

此查询具有分组键。 因此，分组在一起的事件必须被发送到相同事件中心分区。 由于在此示例中我们按 TollBoothID 进行分组，因此应确保在将事件发送到事件中心时，将 TollBoothID 用作分区键。 然后在 Azure 流分析中，可以使用 PARTITION BY PartitionId 继承此分区方案并启用完全并行化。 由于输出是 Blob 存储，因此如要求 #4 所述，无需担心如何配置分区键值。

上一节介绍了一些易并行方案。 本节将介绍不满足实现易并行所需全部要求的方案。

• 输入：具有 8 个分区的事件中心
• 输出：具有 32 个分区的事件中心

如果输入分区数不等于输出分区数，则无论是什么查询，拓扑都不会易并行。 但是，我们仍然可以实现一定程度的并行化。

• 输入：具有 8 个分区的事件中心
• 输出：Power BI

Power BI 输出当前不支持分区。 因此，此方案不易并行。

• 输入：具有 8 个分区的事件中心
• 输出：具有 8 个分区的事件中心
• 兼容性级别：1.0 或 1.1

查询：

    WITH Step1 AS (
    SELECT COUNT(*) AS Count, TollBoothId, PartitionId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1 Partition By TollBoothId
    GROUP BY TumblingWindow(minute, 3), TollBoothId

正如所见，第二步使用 TollBoothId 作为分区键。 此步骤与第一步不同，因此需要执行随机选择。

• 输入：具有 8 个分区的事件中心（未设置“分区键列”，默认为“PartitionId”）
• 输出：具有 8 个分区的事件中心（“分区键列”必须设置为使用“TollBoothId”）
• 兼容性级别 - 1.2 或更高

查询：

    WITH Step1 AS (
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute, 3), TollBoothId

默认情况下，兼容性级别 1.2 或更高级别支持并行查询执行。 例如，只要将“TollBoothId”列设置为输入分区键，就可对上一部分中的查询进行分区。 不需要 PARTITION BY PartitionId 子句。

流分析作业所能使用的流式处理单位总数取决于为作业定义的查询中的步骤数，以及每一步的分区数。

查询可以有一个或多个步骤。 每一步都是由 WITH 关键字定义的子查询。 位于 WITH 关键字外的查询（仅 1 个查询）也计为一步，例如以下查询中的 SELECT 语句：

查询：

    WITH Step1 AS (
        SELECT COUNT(*) AS Count, TollBoothId
        FROM Input1 Partition By PartitionId
        GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )
    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute,3), TollBoothId

此查询有两步。

对步骤进行分区需要下列条件：

• 输入源必须进行分区。
• 查询的 SELECT 语句必须从进行了分区的输入源读取。
• 步骤中的查询必须有 PARTITION BY 关键字。

对查询进行分区后，需在独立的分区组中处理和聚合输入事件，并为每个组生成输出事件。 如果需要对聚合进行组合，则必须创建另一个未分区的步骤来进行聚合。

所有未分区的步骤总共可将一个流分析作业纵向扩展到最多 1 个流单元 (SU V2)。 此外，可以在分区步骤中为每个分区添加 1 个 SU V2。 下表是一些示例。

以下查询计算 3 分钟时段内通过收费站（共 3 个收费亭）的车辆数。 此查询可纵向扩展到 1 个 SU V2。

    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

若要对查询使用更多 SU，必须对输入数据流和查询进行分区。 由于数据流分区设置为 3，因此可将以下经修改的查询纵向扩展到 3 个 SU V2：

    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

对查询进行分区后，会在独立的分区组中处理和聚合输入事件。 此外，还会为每个组生成输出事件。 在输入数据流中，当“分组方式”字段不是分区键时，执行分区可能会导致某些意外的结果。 例如，在前面的查询中，TollBoothId 字段不是 Input1 的分区键。 因此，可以将 TollBooth #1 中的数据分布到多个分区。

流分析会分开处理每个 Input1 分区。 因此，将在相同的翻转窗口为同一收费亭创建多个关于车辆数的记录。 如果不能更改输入分区键，则可通过添加未分区步骤以跨分区聚合值来解决此问题，如下例所示：

    WITH Step1 AS (
        SELECT COUNT(*) AS Count, TollBoothId
        FROM Input1 Partition By PartitionId
        GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute, 3), TollBoothId

此查询可扩展到 4 个 SU V2。

易并行作业是必要的，但并不足以大规模保持较高吞吐量。 每个存储系统及其对应的流分析输出在如何尽量实现最大写入吞吐量的方面各有千秋。 对于任何大规模方案，某些挑战可以通过使用正确的配置得以解决。 本部分讨论几种常见输出的配置，并提供有关保持每秒 1 K、5 K 和 10 K 个事件的引入速率的示例。

以下观测数据使用包含无状态（直通）查询的流分析作业，该查询是一个基本的 JavaScript 用户定义函数 (UDF)，用于写入到事件中心、Azure SQL 或 Azure Cosmos DB。

事件中心解决方案在流单元 (SU) 和吞吐量方面可实现线性缩放，是分析和流式传输流分析数据的最高效方式。 作业可纵向扩展到 66 个 SU V2，这大致相当于处理速率高达 400 MB/秒，即每天可处理 38 万亿个事件。

Azure SQL 支持并行写入（称为继承分区），但默认不会启用此功能。 但是，结合使用继承分区和完全并行查询可能不足以实现更高的吞吐量。 SQL 写入吞吐量在很大程度上取决于数据库配置和表架构。 SQL 输出性能一文详细介绍了可最大程度提高写入吞吐量的参数。 如从 Azure 流分析输出到 Azure SQL 数据库一文中所述，此解决方案无法作为完全并行管道以线性方式扩展到 8 个以上的分区，并且可能需要在 SQL 输出之前重新分区（请参阅 INTO）。 需要使用高级 SKU 来维持较高的 IO 速率，同时，每隔几分钟就会产生日志备份的开销。

流分析的 Azure Cosmos DB 输出已更新为使用兼容性级别 1.2 中的原生集成。 与 1.1 相比，兼容性级别 1.2 明显提高了吞吐量，并减少了 RU 消耗，它是新作业的默认兼容性级别。 该解决方案使用根据 /deviceId 进行了分区的 Azure Cosmos DB 容器，解决方案其余部分采用相同的配置。

所有大规模流式处理 Azure 示例都使用由负载模拟测试客户端提供的事件中心作为输入。 每个输入事件都是一个 1 KB 的 JSON 文档，可轻松地将配置的引入速率转换为吞吐率（1 MB/秒、5 MB/秒和 10 MB/秒）。 事件可模拟 IoT 设备，为最多 1000 台设备发送以下 JSON 数据（简略形式）：

{
    "eventId": "b81d241f-5187-40b0-ab2a-940faf9757c0",
    "complexData": {
        "moreData0": 51.3068118685458,
        "moreData22": 45.34076957651598
    },
    "value": 49.02278128887753,
    "deviceId": "contoso://device-id-1554",
    "type": "CO2",
    "createdAt": "2019-05-16T17:16:40.000003Z"
}

使用 Azure 流分析作业中的“指标”窗格可识别管道中的瓶颈。 查看针对吞吐量的“输入/输出事件”，以及“水印延迟”或“积压事件”，可以确定作业是否跟得上输入速率。 对于事件中心指标，请查看“限制的请求数”并相应地调整阈值单位。 对于 Azure Cosmos DB 指标，请查看“吞吐量”下的“每个分区键范围所使用的最大 RU/秒”，以确保均匀使用分区键范围。 对于 Azure SQL 数据库，请监视“日志 IO”和“CPU”。

如需获取进一步的帮助，可前往 Azure 流分析的 Microsoft 问答页面。

• Azure 流分析简介
• Azure 流分析入门
• Azure 流分析查询语言参考
• Azure 流分析管理 REST API 参考