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

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

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

流分析作业的组成部分有哪些？

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

输入和输出中的分区

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

输入

所有 Azure 流分析 流式处理输入都可以利用分区：事件中心 (Event Hubs)、物联网中心 (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 关键字。

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

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

所有未分区的步骤总共可将一个流分析作业纵向扩展到最多 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

对查询进行分区后，会在独立的分区组中处理和聚合输入事件。 此外，还会为每个组生成输出事件。 在输入数据流中，如果用于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 事件引入速率的示例。

以下观察结果使用包含无状态（直通）查询的流分析作业和能够写入到事件中心、Azure SQL或Azure Cosmos DB的基本 JavaScript 用户定义函数 (UDF)。

事件中心

Event Hubs 解决方案在流单元（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 DB，监视 Log IO 和 CPU。

获取帮助

如需进一步帮助，请尝试访问我们的 Microsoft 问答页面，以获取有关 Azure 流分析 的问题帮助。

后续步骤

• Azure 流分析 介绍
• 开始使用 Azure 流分析
• Azure 流分析查询语言参考
• Azure 流分析 REST API 管理参考