Azure 流分析中的无代码流处理

使用无代码编辑器，无需编写一行代码，即可使用拖放功能轻松开发流分析作业来处理实时流数据。 该体验提供了一个画布，让你可以连接到输入源来快速查看流数据。 然后，可以在将这些数据写入目标之前对其进行转换。

使用无代码编辑器，可以轻松：

• 修改输入架构。
• 执行数据准备操作，如联接和筛选。
• 处理复杂情况，例如用于 group-by 操作的时间窗口聚合（翻转、跳跃和会话窗口）。

创建和运行流分析作业后，可以轻松操作生产工作负荷。 使用一组适当的内置指标进行监视和故障排除。 流分析作业在运行时是按定价模型计费的。

先决条件

在使用无代码编辑器开发流分析作业之前，必须满足以下要求：

• 流分析作业的流式处理输入源和目标资源必须可公开访问，并且不能位于 Azure 虚拟网络中。
• 你必须具有访问流输入和输出资源所需的权限。
• 你必须保留创建和修改 Azure 流分析资源的权限。

Azure 流分析作业

流分析作业是基于三个主要组件构建的：流输入、转换和输出。 你可以拥有任意数量的组件，包括多个输入、具有多个转换的并行分支和多个输出。 有关详细信息，请参阅 Azure 流分析文档。

若要访问用于生成流分析作业的无代码编辑器，有两种方法：

1. 通过 Azure 流分析门户（预览）：创建流分析作业，然后在“概述”页面选的“入门”选项卡中选择“无代码编辑器”，或在左侧面板中选择“无代码编辑器”。

   

2. 通过Azure 事件中心门户：打开事件中心实例。 选择“处理数据”，然后选择任意预定义模板。

   

   预定义模板可帮助您开发和运行作业，以解决各种场景，包括：

   • 使用 Power BI 数据集生成实时仪表板
   • 以 Delta Lake 格式从事件中心捕获数据（预览版）
   • 筛选和引入到 Azure Synapse SQL
   • 将事件中心数据以 Parquet 格式捕获到 Azure Data Lake Storage Gen2 中
   • 在 Azure Cosmos DB 中将数据具体化
   • 筛选并引入到 Azure Data Lake Storage Gen2
   • 扩充数据并引入事件中心
   • 转换数据并将其存储到 Azure SQL 数据库
   • 筛选数据并将其引入 Azure 数据资源管理器

以下屏幕截图显示已完成的流分析作业。 它突出显示了在你创作时可供使用的所有部分。

1. 功能区：在功能区上，各个部分按照经典分析流程的顺序排列：作为输入的事件中心（也称为数据源）、转换（流式提取、转换和加载操作）、输出、一个用于保存进度的按钮和一个用于启动作业的按钮。
2. 图示视图：这是流分析作业的图形表示形式，从输入到操作再到输出。
3. 侧窗格：根据你在图示视图中选择的组件，你会具有修改输入、转换或输出的设置。
4. 用于查看数据预览、创作错误、运行时日志和指标的选项卡：对于每个磁贴，数据预览会向你显示此步骤的结果（对于输入，结果是实时的，对于转换和输出，结果则是按需生成的）。 本部分还汇总了在开发作业时你在作业中可能遇到的任何创作错误或警告。 选择每个错误或警告选择对应的转换。 它还提供作业指标，让你可以监视正在运行的作业的运行状况。

流式处理数据输入

无代码编辑器支持从三类资源中输入流式传输数据：

• Azure 事件中心
• Azure IoT 中心
• Azure Data Lake Storage Gen2

有关流式处理数据输入的详细信息，请参阅将数据作为输入流式传输到流分析中。

作为流输入的 Azure 事件中心

Azure 事件中心是一个大数据流式处理平台和事件引入服务。 它可以每秒接收和处理数百万个事件。 可通过任何实时分析提供程序或批处理/存储适配器转换和存储发送到事件中心的数据。

若要将事件中心配置为作业的输入，请选择“事件中心”图标。 一个磁贴会显示在图示视图中，其中包括用于其配置和连接的侧窗格。

在无代码编辑器中连接到事件中心时，我们建议创建一个新的使用者组（默认选项）。 此方法有助于防止事件中心达到并发读取者的限制。 若要详细了解使用者组以及确定是要选择现有的使用者组还是创建新的使用者组，请参阅使用者组。

如果事件中心位于基本层，则你只能使用现有的 $Default 使用者组。 如果事件中心位于标准或高级层，则你可以创建新的使用者组。

连接到事件中心时，如果选择“托管标识”作为身份验证模式，则会向流分析作业的托管标识授予“Azure 事件中心数据所有者”角色。 若要详细了解事件中心的托管标识，请参阅使用托管标识从 Azure 流分析作业访问事件中心。

托管标识消除了基于用户的身份验证方法的限制。 这些限制包括由于密码更改或每隔 90 天用户令牌过期而需要重新进行身份验证。

设置事件中心的详细信息并选择“连接”后，如果你知道字段名称，可以使用“+ 添加字段”手动添加字段。 若要根据传入消息的示例自动检测字段和数据类型，请选择”自动检测字段”。 选择齿轮符号可根据需要编辑凭据。

当流分析作业检测到这些字段时，你会在列表中看到它们。 还可以在关系图视图下的“数据预览”表中看到传入消息的实时预览。

修改输入数据

始终可以通过选择每个字段旁边的三点符号来编辑字段名称、删除字段、更改数据类型，或更改事件时间（标记为事件时间：如果字段是日期/时间类型，可使用 TIMESTAMP BY 子句）。 还可以从传入消息中展开、选择和编辑任何嵌套字段，如下图所示。

可用数据类型如下：

• DateTime：采用 ISO 格式的日期和时间字段。
• Float：小数。
• Int：整数。
• Record：包含多个记录的嵌套对象。
• String：文本。

Azure IoT 中心作为流式处理输入

Azure IoT 中心是托管在云中的托管服务，充当中央消息中心，用于 IoT 应用程序及其连接设备之间的通信。 发送到 IoT 中心的 IoT 设备数据可用作流分析作业的输入。

若要添加 IoT 中心作为作业的流式输入，请从功能区中选择“输入”下的“IoT 中心”。 然后在右侧面板中填写所需信息，以将 IoT 中心连接到作业。 若要详细了解每个字段的详细信息，请参阅将数据从 IoT 中心流式传输到流分析作业。

Azure Data Lake Storage Gen2 为流式处理输入

Azure Data Lake Storage Gen2 (ADLS Gen2) 是基于云的企业数据湖解决方案。 它旨在可以以任何格式存储大量数据，并促进大数据分析工作负载。 ADLS Gen2 中存储的数据可以通过流分析作为数据流进行处理。 若要详细了解此类输入，请参阅将数据从 ADLS Gen2 流式传输到流分析作业

若要添加 ADLS Gen2 作为作业的流式输入，请从功能区中选择“输入”下的“ADLS Gen2”。 然后在右侧面板中填写所需信息，以将 ADLS Gen2 连接到作业。 若要详细了解每个字段的详细信息，请参阅将数据从 ADLS Gen2 流式传输到流分析作业

引用数据输入

参考数据是静态的，或随时间缓慢变化。 它通常用于扩充传入流以及在作业中执行查找。 例如，可以将数据流输入联接到参考数据，就像执行 SQL 联接以查找静态值一样。 有关参考数据输入的详细信息，请参阅在流分析中使用参考数据进行查找。

无代码编辑器现在支持两个参考数据源：

• Azure Data Lake Storage Gen2
• Azure SQL 数据库

Azure Data Lake Storage Gen2 作为参考数据

参考数据建模为 blob 序列，这些 blob 按 blob 名称中指定的日期/时间组合顺序升序排列。 只能使用大于序列中最后一个 blob 指定的日期/时间的日期/时间将 blob 添加到序列的末尾。 blob 是在输入配置中定义的。

首先，在功能区的“输入”部分下，选择“参考 ADLS Gen2”。 若要查看有关每个字段的详细信息，请参阅在流分析中使用参考数据进行查找中有关 Azure Blob 存储的部分。

然后上传一个 JSON 数组文件。 将检测该文件中的字段。 使用此参考数据对来自事件中心的流输入数据执行转换。

Azure SQL 数据库作为参考数据

可以在无代码编辑器中使用 Azure SQL 数据库作为流分析作业的参考数据。 有关详细信息，请参阅在流分析中使用参考数据进行查找中有关 SQL 数据库的部分。

若要将 SQL 数据库配置为参考数据输入，请在功能区的“输入”部分下选择“参考 SQL 数据库”。 然后填写信息以连接参考数据库，并选择包含所需列的表。 还可以通过手动编辑 SQL 查询从表中提取参考数据。

转换

流数据转换本质上与批处理数据转换不同。 几乎所有流数据都有一个时间组件，它会影响所涉及的任何数据准备任务。

若要向作业添加流数据转换，请在功能区的“操作”部分下，选择该转换对应的转换符号。 相应的磁贴将放在图示视图中。 选择转换后，你将看到该转换的侧窗格以进行配置。

筛选器

使用“筛选器”转换以基于输入中的字段值筛选事件。 根据数据类型（数值或文本），转换将保留与所选条件匹配的值。

管理字段

利用“管理字段”转换，你可以添加、删除或重命名从输入或其他转换中传入的字段。 使用侧窗格上的设置可以选择添加一个新的字段，方法是选择“添加字段”或一次添加所有字段。

还可以使用“内置函数”添加新字段以聚合来自上游的数据。 目前支持的内置函数是“字符串函数”、“日期和时间函数”、“数学函数”中的某些函数。 若要详细了解这些函数的定义，请参阅内置函数（Azure 流分析）。

聚合

你可以使用“聚合”转换来计算一段时间内每次新事件发生时的聚合（总和、最小值、最大值或平均值）。 此操作还允许你根据数据中的其他维度对聚合进行筛选或切片。 在同一转换中可以有一个或多个聚合。

若要添加聚合，请选择转换符号。 然后连接输入，选择聚合，添加任何筛选或切片维度，并选择将计算聚合的时间范围。 在此示例中，我们将按车辆来自的省/市/自治区计算过去 10 秒的通行费值总和。

若要将另一个聚合添加到相同转换，请选择“添加聚合函数”。 请记住，筛选或切片将应用于转换中的所有聚合。

联接

可以使用“联接”转换根据所选的字段对合并两个输入中的事件。 如果不选择字段对，默认情况下连接将基于时间。 默认设置使此转换不同于批处理转换。

与常规联接一样，对于联接逻辑，还可以使用选项：

• 内部联接：只包括两个表（其中字段对匹配）中的记录。 在此示例中，牌照同时匹配两个输入。
• 左外部联接：包含左侧（第一个）表中的所有记录，并且仅包含第二个表中与字段对匹配的记录。 如果没有匹配项，则第二个输入中的字段将为空。

若要选择联接类型，请在侧窗格中选择首选类型的符号。

最后，选择你希望计算哪个时间段的联接。 在此示例中，将计算过去 10 秒的联接。 请记住，时间越长，输出的频率越低，用于转换的处理资源就越多。

默认情况下，将包括两个表中的所有字段。 输出中的左前缀（第一个节点）和右前缀（第二个节点）有助于区分源。

Group by

使用“分组依据”转换计算某个时间窗口内所有事件的聚合。 可以按一个或多个字段中的值进行分组。 它类似于“聚合”转换，但提供了更多聚合选项。 它还包括时间窗口的更复杂选项。 同样，与“聚合”类似，你可以为每个转换添加多个聚合。

转换中可用的聚合如下：

• 平均值
• Count
• 最大值
• 最低
• 百分位（连续和离散）
• 标准偏差
• Sum
• Variance

若要配置转换，请执行以下操作：

1. 选择首选聚合。
2. 选择要聚合的字段。
3. 如果要在另一个维度或类别上获取聚合计算，请选择可选的“分组依据”字段。 例如：“省/市/自治区”。
4. 选择时间窗口函数。

若要将另一个聚合添加到相同转换，请选择“添加聚合函数”。 请记住，“分组依据”字段和窗口函数将应用于转换中的所有聚合。

时间窗口结束时间戳显示为转换输出的一部分以供参考。 有关流分析作业支持的时间窗口的详细信息，请参阅开窗函数（Azure 流分析）。

Union

使用“联合”转换可以连接两个或多个输入，将具有共享字段（具有相同名称和数据类型）的事件添加到一个表中。 不匹配的字段将被删除，不会包含在输出中。

扩展数组

使用“扩展数组”转换可为数组中的每个值创建一个新行。

流输出

无代码拖放体验目前支持多个输出接收器来存储已处理的实时数据。

Azure Data Lake Storage Gen2

Data Lake Storage Gen2 使 Azure 存储成为在 Azure 上构建企业 Data Lake 的基础。 它设计为处理多个 PB 量级的信息，同时保持数百 GB 的吞吐量。 它让你可以轻松管理海量数据。 Azure Blob 存储提供了一种经济高效且可缩放的解决方案，用于在云中存储大量非结构化数据。

在功能区的“输出”部分下，选择“ADLS Gen2”作为流分析作业的输出。 然后选择要将作业输出发送到的容器。 有关流分析作业的 Azure Data Lake Gen2 输出的详细信息，请参阅来自 Azure 流分析的 Blob 存储和 Azure Data Lake Gen2 输出。

连接到 Azure Data Lake Storage Gen2 时，如果选择“托管标识”作为身份验证模式，则会向流分析作业的托管标识授予“存储 Blob 数据参与者”角色。 若要详细了解 Azure Data Lake Storage Gen2 的托管标识，请参阅使用托管标识对 Azure Blob 存储的 Azure 流分析作业进行身份验证。

托管标识消除了基于用户的身份验证方法的限制。 这些限制包括由于密码更改或每隔 90 天用户令牌过期而需要重新进行身份验证。

由于没有代码编辑器输出，ADLS Gen2 支持“恰好一次”交付（预览版）。 可以在 ADLS Gen2 配置的“写入模式”部分启用它。 有关此功能的详细信息，请参阅 Azure Data Lake Gen2 中的“恰好一次”交付（预览版）

ADLS Gen2 支持“写入 Delta Lake 表(预览版)”，因为没有代码编辑器输出。 可以在 ADLS Gen2 配置中的“序列化”部分访问此选项。 有关此功能的详细信息，请参阅写入 Delta Lake 表。

Azure Synapse Analytics

Azure 流分析作业可将输出发送到 Azure Synapse Analytics 中的专用 SQL 池表，并可以处理高达每秒 200 MB 的吞吐率。 流分析支持报告和仪表板等工作负载的最苛刻实时分析和热路径数据处理需求。

在功能区的“输出”部分下，选择“Synapse”作为流分析作业的输出。 然后选择要将作业输出发送到的 SQL 池表。 有关流分析作业的 Azure Synapse 输出的详细信息，请参阅来自 Azure 流分析的 Azure Synapse Analytics 输出。

Azure Cosmos DB

Azure Cosmos DB 是全球分布式数据库服务，可在全球范围内提供无限制的弹性缩放。 它还针对架构无关的数据模型提供丰富的查询和自动索引。

在功能区的“输出”部分下，选择“CosmosDB”作为流分析作业的输出。 有关流分析作业的 Azure Cosmos DB 输出的详细信息，请参阅来自 Azure 流分析的 Azure Cosmos DB 输出。

连接到 Azure Cosmos DB 时，如果选择“托管标识”作为身份验证模式，则会向流分析作业的托管标识授予“参与者”角色。 若要详细了解 Azure Cosmos DB 的托管标识，请参阅使用托管标识从 Azure 流分析作业访问 Azure Cosmos DB（预览版）。

无代码编辑器中的 Azure Cosmos DB 输出也支持托管标识身份验证方法，其优点与前面的 ADLS Gen2 输出中所述相同。

Azure SQL 数据库

Azure SQL 数据库是完全托管的平台即服务 (PaaS) 数据库引擎，可帮助你为 Azure 中的应用程序和解决方案创建高度可用的高性能数据存储层。 使用无代码编辑器，可以配置 Azure 流分析作业以将处理后的数据写入 SQL 数据库中的现有表。

若要将 Azure SQL 数据库配置为输出，请在功能区的“输出”部分下选择“SQL 数据库”。 然后填写所需的信息以连接 SQL 数据库，并选择要将数据写入到的表。

有关流分析作业的 Azure SQL 数据库输出的详细信息，请参阅来自 Azure 流分析的 Azure SQL 数据库输出。

事件中心

当实时数据传入 ASA 时，无代码编辑器可以转换、扩充数据，然后将数据输出到另一个事件中心。 可以在配置 Azure 流分析作业时选择“事件中心”输出。

若要将事件中心配置为输出，请在功能区的“输出”部分下选择“事件中心”。 然后填写所需的信息以连接要将数据写入到的事件中心。

有关流分析作业的事件中心输出的详细信息，请参阅来自 Azure 流分析的事件中心输出。

Azure 数据资源管理器

Azure 数据资源管理器是完全托管的高性能、大数据分析平台，让你可以轻松分析大量数据。 也可以使用无代码编辑器将“Azure 数据资源管理器”用作 Azure 流分析作业的输出。

若要将 Azure 数据资源管理器配置为输出，请在功能区的“输出”部分下选择“Azure 数据资源管理器”。 然后填写所需的信息以连接 Azure 数据资源管理器数据库，并指定要将数据写入到的表。

有关流分析作业的 Azure 数据资源管理器输出的详细信息，请参阅来自 Azure 流分析的 Azure 数据资源管理器输出（预览版）。

Power BI

Power BI 为数据分析结果提供了全面的可视化体验。 将 Power BI 输出到流分析后，已处理的流数据将写入 Power BI 流数据集，然后可用于生成准实时的 Power BI 仪表板。 若要详细了解如何生成准实时仪表板，请参阅使用从流分析（无代码编辑器）生成的 Power BI 数据集生成实时仪表板。

若要将 Power BI 配置为输出，请在功能区上的“输出”部分下选择“Power BI”。 然后填写所需的信息以连接 Power BI 工作区，并提供要将数据写入到的流式处理数据集和表的名称。 若要详细了解每个字段的详细信息，请参阅 Azure 流分析的 Power BI 输出。

数据预览、创作错误、运行时日志和指标

无代码拖放体验提供的工具可帮助你创作流数据分析管道、排查其故障，并对其性能进行评估。

输入的实时数据预览

连接到事件中心并在图示视图中选择相应的磁贴（“数据预览”选项卡）时，如果以下所有条件都成立，可实时预览传入数据：

• 正在推送数据。
• 输入配置正确。
• 已添加字段。

如以下屏幕截图所示，如果想查看或向下钻取特定内容，可以暂停预览 (1)。 或者，如果已完成，可以再次启动它。

还可查看特定记录（表中的“单元格”）的详细信息，具体做法是选择该记录，然后选择“显示/隐藏详细信息” (2)。 屏幕截图显示了记录中嵌套对象的详细视图。

转换和输出的静态预览

在图示视图中添加并设置任何步骤后，可以选择“获取静态预览”测试其行为。

在你执行后，流分析作业会评估所有转换和输出，以确保它们已正确配置。 然后，流分析会在静态数据预览中显示结果，如下图所示。

可以通过选择“刷新静态预览”来刷新预览 (1)。 当你刷新预览时，流分析作业会从输入获取新数据并评估所有转换。 然后，它会再次使用你可能已执行的任何更新发送输出。 “显示/隐藏详细信息”选项也可用 (2)。

创作错误

如果有任何创作错误或警告，“创作错误”选项卡将列出它们，如以下屏幕截图所示。 该列表包含有关错误或警告的详细信息、卡片类型（输入、转换或输出）、错误级别以及错误或警告说明。

运行时日志

当作业运行时，将显示警告、错误或信息级别的运行时日志。 如果你要编辑流分析作业拓扑或配置以用于故障排除，这些日志非常有用。 我们强烈建议在“设置”中打开诊断日志并将其发送到 Log Analytics 工作区，以便更深入地查看正在运行的作业以进行调试。

在以下屏幕截图示例中，用户已使用一个与作业输出字段不匹配的表架构配置了 SQL 数据库输出。

指标

如果作业正在运行，你可以在“指标”选项卡上监视作业的运行状况。默认显示的四个指标是“水印延迟”、“输入事件数”、“积压的输入事件数”和“输出事件数”。 可以使用这些指标来了解事件是否流入和流出作业且未产生任何输入积压工作。

可以从列表中选择更多指标。 若要详细了解所有指标，请参阅 Azure 流分析作业指标。

启动流分析作业

在创建作业时，可以随时保存作业。 为作业配置流式输入、转换及流式输出后，可以启动作业。

可配置以下选项：

• 输出开始时间：启动作业时，可以选择该作业开始创建输出的时间。
  • 立即：此选项使输出事件流的起点与作业启动时间相同。
  • 自定义：可以选择输出的起点。
  • 上次停止时：如果作业先前已启动，但被手动停止或失败，则此选项将可供使用。 选择此选项时，上次输出时间将用来重启作业，因此不会丢失数据。
• 流单元：流单元 (SU) 表示在运行时分配给作业的计算和内存量。 如果你不确定要选择多少 SU，我们建议你从 3 开始，并根据需要进行调整。
• 输出数据错误处理：仅当流分析作业生成的输出事件不符合目标接收器的架构时，输出数据错误处理的策略才适用。 可以选择“重试”或“丢弃”来配置此策略。 有关详细信息，请参阅 Azure 流分析输出错误策略。
• 启动：此按钮用于启动流分析作业。

Azure 事件中心门户中的流分析作业列表

若要在 Azure 事件中心门户中查看使用无代码拖放体验来创建的所有流分析作业的列表，请选择“处理数据”>“流分析作业”。

下面是“流分析作业”选项卡的元素：

• 筛选：可以按作业名称筛选列表。
• 刷新：列表目前不会自动刷新。 使用“刷新”按钮可刷新列表并查看最新状态。
• 作业名称：此区域中的名称是在第一个作业创建步骤中提供的名称。 你无法编辑它。 选择作业名称可在无代码拖放体验中打开作业，在该体验中可以停止作业、编辑作业，以及再次启动作业。
• 状态：此区域显示作业的状态。 选择列表顶部的“刷新”可查看最新状态。
• 流单元：此区域显示启动作业时选择的流单元的数量。
• 输出水印：此区域提供作业生成的数据的活动性指示器。 时间戳之前的所有事件都已计算过。
• 作业监视：选择“打开指标”可查看与此流分析作业相关的指标。 有关可用于监视流分析作业的指标的详细信息，请参阅 Azure 流分析作业指标。
• 操作：启动、停止或删除作业。

后续步骤

了解如何使用无代码编辑器通过预定义模板处理常见方案：

• Azure 流分析简介
• 使用 Azure 门户监视流分析作业
• 理解 Azure 流分析的输入
• Azure 流分析的输出