使用 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 输出当前不支持分区。 因此，此方案不易并行。

使用不同 PARTITION BY 值进行多步骤查询

• 输入：具有 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 作为分区键。 此步骤与第一步不同，因此需要随机进行。

使用不同 PARTITION BY 值进行多步骤查询

• 输入：具有 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 关键字。

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

计算作业的最大流式处理单位数

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

缩放示例

以下查询计算 3 分钟时段内通过收费站（共 3 个收费亭）的车辆数。 您可以将此查询扩展至一个 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

对查询进行分区时，会在单独的分区组中处理和聚合输入事件。 该查询为每个组生成输出事件。 在输入数据流中，当GROUP BY字段不是分区键时，进行分区可能会导致某些意外的结果。 例如，在前面的查询中，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

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

Azure Cosmos DB

流分析的 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 数据库，请监控“日志输入/输出”和“处理器”。

获取帮助

如需进一步的帮助，请尝试 Azure 流分析的Microsoft问答页面。

后续步骤

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