将 SQL 数据库中的参考数据用于 Azure 流分析作业

参考数据是一个静态或缓慢变化的数据集，你会将其与流式数据结合起来丰富，比如在销售事件流中添加产品细节。 Azure 流分析 支持 Azure SQL 数据库 作为参考数据来源，因此你可以查找并将这些数据与实时输入结合。

本文展示了如何通过使用 Azure 门户和 Visual Studio 配合流分析工具，配置 Azure SQL 数据库 作为 Stream Analytics 作业的参考数据输入。

通过使用 Azure 门户添加 SQL 数据库参考数据

请使用以下步骤，通过使用 Azure 门户添加 Azure SQL 数据库 作为参考输入源：

门户先决条件

1. 创建流分析作业。

2. 为流分析作业创建一个存储账户。

   重要

   Azure 流分析 会在该存储账户中保留快照。 在配置保留策略时，确保所选时间段包含你想要的Stream Analytics任务所需的恢复时间。

3. 创建包含流分析作业用作参考数据的数据集的 Azure SQL 数据库。

定义 SQL 数据库参考数据输入

1. 在流分析作业中，选择 输入（位于 作业拓扑 下）。 选择 添加引用输入，然后选择 SQL数据库。

   

2. 填写Stream Analytics输入配置。 选择数据库名称、服务器名称和登录凭证。 要定期刷新参考数据输入，请选择 “开启 ”，并在 DD：HH：MM 中指定刷新率。 对于刷新率较短的大型数据集，delta查询通过检索SQL数据库中在开始时间 @deltaStartTime和结束时间 @deltaEndTime之间插入或删除的所有行来跟踪参考数据的变化。

   有关详细信息，请参阅 增量查询。

   

3. 在 SQL 查询编辑器中测试快照查询。 更多信息请参见使用 Azure 门户的 SQL 查询编辑器连接和查询数据。

在作业配置中指定存储账户

进入配置中的存储账户设置，然后选择添加存储账户。

启动作业

1. 配置好其他输入、输出和查询后，开始流分析任务。

通过使用 Visual Studio 添加 SQL 数据库参考数据

使用Visual Studio，请按照以下步骤添加Azure SQL 数据库作为参考输入源：

Visual Studio 先决条件

1. 安装用于 Visual Studio 的流分析工具。 流分析工具支持以下版本的Visual Studio：

   • Visual Studio 2015
   • Visual Studio 2019

2. 通过用于 Visual Studio 的流分析工具快速入门来熟悉工具。

3. 创建存储帐户。

   重要

   Azure 流分析 会在该存储账户中保留快照。 在配置保留策略时，确保所选时间段包含你想要的Stream Analytics任务所需的恢复时间。

创建 SQL 数据库表

使用 SQL Server Management Studio 创建用于存储参考数据的表。 有关详细信息，请参阅使用 SSMS 设计第一个 Azure SQL 数据库。

以下陈述构成示例表：

create table chemicals(Id Bigint,Name Nvarchar(max),FullName Nvarchar(max));

选择订阅方案

1. 在 Visual Studio 中，在“视图”菜单中选择“服务器资源管理器”。

2. 选择并长按（或右键点击）Azure，选择连接 Microsoft Azure 订阅，然后用你的 Azure 账户登录。

创建流分析项目

1. 选择文件>新建项目。

2. 在模板列表中，选择 Stream Analytics，然后选择 Azure 流分析 Application。

3. 输入项目 名称、 地点和 解决方案名称，然后选择 确定。

   

定义 SQL 数据库参考数据输入

1. 创建新输入。

   

2. 在解决方案资源管理器中打开Input.json。

3. 填写“流分析输入配置”。 输入数据库名称、服务器名称、刷新类型和刷新率。 以 DD:HH:MM 格式指定刷新频率。

   

   如果你选择只执行一次或周期性执行，Visual Studio会在项目中的 Input.json 文件节点下生成一个名为[Input Alias].snapshot.sql的SQL代码背后文件。

   

   如果你选择使用 Delta 定期刷新，Visual Studio会生成两个 SQL CodeBehind 文件：[Input Alias].snapshot.sql[Input Alias].delta.sql。

   

4. 在编辑器中打开 SQL 文件并编写 SQL 查询。

5. 如果你用的是Visual Studio 2019并且安装了SQL Server Data Tools，可以通过选择执行来测试查询。 会打开一个向导帮助你连接 SQL 数据库，查询结果会显示在底部的窗口中。

指定存储帐户

打开 JobConfig.json 指定存储账户以存储SQL引用快照。

在本地进行测试并部署到 Azure

在你将作业部署到 Azure 之前，可以在本地对实时输入数据测试查询逻辑。 有关此功能的更多信息，请参见 Visual Studio 的 Azure 流分析 工具在本地测试实时数据（预览版）。 测试结束后，选择提交到 Azure。 若要了解如何创建该作业，请参阅使用适用于 Visual Studio 的 Azure 流分析 工具创建 Stream Analytics 作业快速入门。

增量查询

使用delta查询时，使用Azure SQL 数据库中的时序表。

1. 在 Azure SQL 数据库中创建时态表。

      CREATE TABLE DeviceTemporal
      (
         [DeviceId] int NOT NULL PRIMARY KEY CLUSTERED
         , [GroupDeviceId] nvarchar(100) NOT NULL
         , [Description] nvarchar(100) NOT NULL
         , [ValidFrom] datetime2 (0) GENERATED ALWAYS AS ROW START
         , [ValidTo] datetime2 (0) GENERATED ALWAYS AS ROW END
         , PERIOD FOR SYSTEM_TIME (ValidFrom, ValidTo)
      )
      WITH (SYSTEM_VERSIONING = ON (HISTORY_TABLE = dbo.DeviceHistory));  -- DeviceHistory table will be used in Delta query
   

2. 创作快照查询。

   使用 @snapshotTime 参数指示流分析运行时从系统时间有效的SQL数据库时序表获取参考数据集。 如果你不提供这个参数，可能会因为时钟偏斜而获得不准确的基准参考数据集。 以下示例展示了完整的快照查询：

      SELECT DeviceId, GroupDeviceId, [Description]
      FROM dbo.DeviceTemporal
      FOR SYSTEM_TIME AS OF @snapshotTime
   

3. 编写增量查询。

   该查询检索 SQL 数据库中在开始时间（即 @deltaStartTime）与结束时间（即 @deltaEndTime）之间插入或删除的所有行。 增量查询必须返回与快照查询以及列操作相同的列。 该列定义了行在 @deltaStartTime 和 @deltaEndTime之间是插入还是删除。 如果插入了记录，则生成的行将标记为 1；如果删除了记录，则标记为 2。 查询还必须从 SQL Server 端添加 水印，以确保正确地捕获增量周期内的所有更新。 使用不含 水印 的Delta查询可能导致引用数据集错误。

   对于已更新的记录，时态表会通过记录一次插入操作和一次删除操作来进行记录。 流分析运行时会将delta查询的结果应用到上一个快照，以保持参考数据的更新。 以下示例显示了一个增量查询：

      SELECT DeviceId, GroupDeviceId, Description, ValidFrom as _watermark_, 1 as _operation_
      FROM dbo.DeviceTemporal
      WHERE ValidFrom BETWEEN @deltaStartTime AND @deltaEndTime   -- records inserted
      UNION
      SELECT DeviceId, GroupDeviceId, Description, ValidTo as _watermark_, 2 as _operation_
      FROM dbo.DeviceHistory   -- table we created in step 1
      WHERE ValidTo BETWEEN @deltaStartTime AND @deltaEndTime     -- record deleted
   

   除增量查询外，流分析运行时还可能定期运行快照查询以存储检查点。

   重要

   使用参考数据差量查询时，不要多次对时间参考数据表进行相同的更新。 这可能导致错误的结果。 这里有一个可能导致参考数据产生错误结果的例子：

    UPDATE myTable SET VALUE=2 WHERE ID = 1;
    UPDATE myTable SET VALUE=2 WHERE ID = 1;
   

   正确示例：

    UPDATE myTable SET VALUE = 2 WHERE ID = 1 and not exists (select * from myTable where ID = 1 and value = 2);
   

   该条件确保不会有重复更新。

测试您的查询

确认你的查询返回的是Stream Analytics职位用作参考数据的预期数据集。 要测试你的查询，请在门户的“职位拓扑”部分进入输入。 然后在你的SQL数据库参考输入中选择 “样本数据 ”。 样本可用后，你可以下载文件，检查返回的数据是否符合预期。 为了优化你的开发和测试迭代，可以使用 Visual Studio 的 Stream Analytics 工具。 你也可以使用其他你喜欢的工具，先确保查询从你的 Azure SQL 数据库 返回正确的结果，然后在你的 Stream Analytics 作业中使用该查询。

使用 Visual Studio Code 测试查询

请在 Visual Studio Code 上安装 Azure 流分析工具和 SQL Server (mssql) 并设置 ASA 项目。 有关详细信息，请参阅快速入门：在 Visual Studio Code 中创建 Azure 流分析作业和 SQL Server (mssql) 扩展教程。

1. 配置您的 SQL 参考数据输入源。

   

2. 选择SQL Server图标，选择添加连接。

   

3. 填写连接信息。

   

4. 选择并长按（或右键点击）参考 SQL，然后选择 执行查询。

   

5. 选择连接。

   

6. 查看并验证查询结果。

   

常见问题

在 Azure 流分析 中使用 SQL 参考数据输入会增加费用吗？

流分析作业中没有额外的每个流式处理单元成本。 但是，流分析作业必须有一个关联的 Azure 存储帐户。 Stream Analytics 作业在作业启动和刷新期间查询 SQL 数据库以获取参考数据集，并将该快照存储在存储账户中。 存储这些快照会产生额外费用，具体费用详见 Azure 存储账户的定价页面。

我如何知道参考数据快照是否被从 SQL 数据库查询并用于 Azure 流分析 作业？

两个指标按逻辑名称（Azure门户中的指标）筛选，可以让你监控SQL数据库参考数据输入的健康状况。

• InputEvents：该指标衡量从 SQL 数据库参考数据集加载的记录数量。
• InputEventBytes：此指标度量流分析作业内存中载入的参考数据快照大小。

这两个指标共同表明作业是否查询SQL数据库获取参考数据集，然后将其加载到内存。

我需要特殊类型的 Azure SQL 数据库 吗？

Azure 流分析 可以支持任何类型的 Azure SQL 数据库。 不过，你为参考数据输入设置的刷新率可能会影响你的查询负载。 若要使用增量查询选项，请在 Azure SQL 数据库中使用时态表。

为什么 Azure 流分析 会在 Azure 存储 账户中存储快照？

流分析保证恰好一次事件处理和至少一次事件传递。 如果暂时性问题影响到你的工作，需要少量重放来恢复状态。 要启用重放，这些快照必须存储在 Azure 存储 账户中。 有关检查点重播的更多信息，请参见 Azure 流分析 jobs 中的检查点和重播概念。

相关内容

• 了解 Azure 流分析的输出
• Azure 流分析输出到 Azure SQL 数据库
• 通过 Azure 流分析提高 Azure SQL 数据库的吞吐量性能
• 使用托管标识从 Azure 流分析作业访问 Azure SQL 数据库或 Azure Synapse Analytics
• 使用 Azure Functions 更新或合并 Azure SQL 数据库中的记录
• 快速入门：使用 Azure 门户创建流分析作业