你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

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

项目
11/17/2023

Azure 流分析支持将 Azure SQL 数据库用作参考数据的输入源。可以在 Azure 门户和 Visual Studio 中配合流分析工具将 SQL 数据库用作流分析作业的参考数据。本文演示如何使用这两种方法。

Azure 门户

在 Azure 门户中使用以下步骤将 Azure SQL 数据库添加为参考输入源：

门户先决条件

创建流分析作业。
创建流分析作业使用的存储帐户。

重要

Azure 流分析会在此存储帐户中保留快照。在配置保留策略时，必须确保所选时间跨度有效包含流分析作业所需的恢复持续时间。
创建包含数据集的 Azure SQL 数据库，流分析作业将使用该数据集作为参考数据。

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

在流分析作业中，选择“作业拓扑”下的“输入”。单击“添加参考输入”并选择“SQL 数据库”。
填写“流分析输入配置”。选择数据库名称、服务器名称、用户名和密码。如果希望参考数据输入定期刷新，请选择“打开”以指定刷新频率（采用 DD:HH:MM 格式）。如果你有刷新频率很快的大数据集。通过增量查询，你可以检索从开始时间 @deltaStartTime 到结束时间 @deltaEndTime 期间 SQL 数据库中插入或删除的所有行，从而跟踪参考数据中的更改。

请参阅增量查询。

When SQL Database is selected, the SQL Database New input page appears. There is a configuration form in the left pane, and a Snapshot query in the right pane.

在 SQL 查询编辑器中测试快照查询。有关详细信息，请参阅使用 Azure 门户的 SQL 查询编辑器进行连接并查询数据

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

导航到“配置”下的“存储帐户设置”，然后选择“添加存储帐户”。

Storage account settings is selected in the left pane. There is an Add storage account button in the right pane.

启动作业

配置其他输入、输出和查询后，可以启动流分析作业。

适用于 Visual Studio 的工具

在 Visual Studio 中使用以下步骤将 Azure SQL 数据库添加为参考输入源：

Visual Studio 先决条件

安装用于 Visual Studio 的流分析工具。支持以下 Visual Studio 版本：
- Visual Studio 2015
- Visual Studio 2019
通过用于 Visual Studio 的流分析工具快速入门来熟悉工具。
创建存储帐户。

重要

Azure 流分析会在此存储帐户中保留快照。在配置保留策略时，必须确保所选时间跨度有效包含流分析作业所需的恢复持续时间。

创建 SQL 数据库表

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

以下示例中使用的示例表是通过以下语句创建的：

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

选择自己的订阅

在 Visual Studio 中，在“视图”菜单中选择“服务器资源管理器”。
右键单击“Azure”，选择“连接到 Microsoft Azure 订阅”，然后使用 Azure 帐户登录。

创建流分析项目

选择“文件”>“新建项目”。
在左侧的模板列表中，选择“流分析”，然后选择“Azure 流分析应用程序”。
输入项目的“名称”、“位置”和“解决方案名称”，然后选择“确定”。

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

创建新输入。
在“解决方案资源管理器”中双击“Input.json”。
填写“流分析输入配置”。选择数据库名称、服务器名称、刷新类型和刷新频率。以 DD:HH:MM 格式指定刷新频率。

如果选择“仅执行一次”或“定期执行”，项目中的“Input.json”文件节点下会生成一个名为 [Input Alias].snapshot.sql 的 SQL CodeBehind 文件。

如果选择“使用增量数据定期刷新”，则会生成两个 SQL CodeBehind 文件： [Input Alias].snapshot.sql 和 [Input Alias].delta.sql。
在编辑器中打开 SQL 文件并编写 SQL 查询。
如果使用 Visual Studio 2019 并且已安装 SQL Server Data Tools，可以单击“执行”来测试查询。此时会弹出一个可帮助你连接到 SQL 数据库的向导窗口，查询结果会出现在窗口底部。

指定存储帐户

打开“JobConfig.json”以指定用于存储 SQL 参考快照的存储帐户。

Stream Analytics Job Configure Configuration is shown with default values. The Global Storage Settings are highlighted.

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

将作业部署到 Azure 之前，可在本地针对实时输入数据测试查询逻辑。有关此功能的详细信息，请参阅使用用于 Visual Studio 的 Azure 流分析工具在本地测试实时数据（预览）。完成测试后，单击“提交到 Azure”。请参考使用用于 Visual Studio 的 Azure 流分析工具创建流分析快速入门来了解如何启动作业。

增量查询

使用增量查询时，建议使用 Azure SQL 数据库中的时态表。

在 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

创作快照查询。

请使用 @snapshotTime 参数来指示流分析运行时从在当前系统时间有效的 SQL 数据库时态表中获取参考数据集。如果不提供此参数，可能会出于时钟偏差的原因获取不准确的基本参考数据集。完整的快照查询示例如下所示：
```
   SELECT DeviceId, GroupDeviceId, [Description]
   FROM dbo.DeviceTemporal
   FOR SYSTEM_TIME AS OF @snapshotTime
```
创作增量查询。

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

对于更新的记录，时态表将通过捕获插入和删除操作来执行簿记。然后，流分析运行时将增量查询的结果应用到前一快照，以保持参考数据的最新状态。下面显示了增量查询的示例：
```
   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);
```
这可确保不会执行重复更新。

测试查询

请务必验证查询是否返回流分析作业将用作参考数据的所需数据集。若要测试查询，请转到门户上“作业拓扑”部分下的“输入”。然后，可以在“SQL 数据库参考输入”中选择“示例数据”。示例可用后，可以下载该文件并检查返回的数据是否符合预期。如果希望优化开发和测试迭代，建议使用适用于 Visual Studio 的流分析工具。也可以使用你喜欢的任何其他工具，首先确保查询从 Azure SQL 数据库返回正确的结果，然后在流分析作业中使用该查询。