更新策略概述

适用于：✅Azure 数据资源管理器

更新策略是在将新数据写入表时触发的自动化机制。 更新策通过运行查询来转换引入的数据，无需特殊的业务流程，并将结果保存到目标表。 可以在单个表上定义多个更新策略，从而进行不同的转换并将数据同时保存到多个表。 目标表可以具有与源表不同的架构、保留策略和其他策略。

例如，高速率跟踪源表可以包含格式化为自由文本列的数据。 目标表可以包含特定类型的跟踪行，以及使用 parse 运算符转换源表的自由文本数据生成的结构良好的架构。 有关详细信息，请参阅常见方案。

下图描绘了更新策略的概要视图。 它显示了在将数据添加到第二个源表时触发的两个更新策略。 触发后，转换的数据将添加到两个目标表中。

更新策略查询

如果更新策略是在目标表上定义的，则可以对引入源表的数据运行多个查询。 如果有多个更新策略，则执行顺序不一定是已知的。

查询限制

在策略的 Source 部分中或由 Query 部分引用的函数中引用 Query 表时：

更新策略对象

一个表可以具有零个或多个与之关联的更新策略对象。 每一个此类对象都表示为 JSON 属性包，并定义了以下属性。

管理命令

更新策略管理命令包括：

• .show table *TableName* policy update 显示表的当前更新策略。
• .alter table *TableName* policy update 定义表的当前更新策略。
• .alter-merge table *TableName* policy update 将定义追加到表的当前更新策略。
• .delete table *TableName* policy update 删除表的当前更新策略。

更新策略在引入后启动

将数据引入或移动到源表，或在源表中创建扩展时，更新策略会生效。 这些操作可以通过以下任意命令完成：

• .ingest （pull）
• .ingest （内联）
• .set | .append | .set-or-append | .set-or-replace
• .move 盘区
• .replace extents
  • PropagateIngestionProperties 命令仅在引入操作中生效。 当更新策略作为 .move extents 或 .replace extents 命令的一部分触发时，此选项无效。

使用源表通配符模式更新策略

更新策略支持从共享相同模式的多个源表引入，同时使用与更新策略查询相同的查询。 如果有多个源表，通常共享同一架构（或共享通用架构的列的子集），并且想要在引入到其中任一表时触发引入到单个目标表，这非常有用。 在这种情况下，可以定义具有通配符 Source的单个更新策略，而不是为单个源表定义多个更新策略。 Query 更新策略必须符合与模式匹配的所有源表。 若要在更新策略查询中引用源表，可以使用名为 <a0/> 的特殊符号。 请参阅 通配符更新策略示例中的示例。

从源表中删除数据

将数据引入目标表后，可以选择将其从源表中删除。 请在源表的0sec中将软删除时间段设置义为 00:00:00（或 ），并将更新策略设置为“事务性”。 下列条件适用：

• 无法从源表查询源数据
• 引入操作期间，源数据不会永久保存到持久性存储中
• 操作性能提高。 在源表的分片上，为后台清理操作减少了引入后的资源。

性能影响

更新策略可能会影响性能，并且数据盘区的引入将乘以目标表的数量。 优化与策略相关的查询非常重要。 可以在创建或更改策略之前或在查询中使用的函数上，调用现有分片上的策略来测试更新策略的性能影响。

评估资源使用情况

将 .show queries 与以下参数结合使用来评估资源使用情况（CPU、内存等）：

• 将 Source 属性（源表名称）设置为 MySourceTable
• 将 Query 属性设置为调用名称为 MyFunction() 的函数

// '_extentId' is the ID of a recently created extent, that likely hasn't been merged yet.
let _extentId = toscalar(
    MySourceTable
    | project ExtentId = extent_id(), IngestionTime = ingestion_time()
    | where IngestionTime > ago(10m)
    | top 1 by IngestionTime desc
    | project ExtentId
);
// This scopes the source table to the single recent extent.
let MySourceTable =
    MySourceTable
    | where ingestion_time() > ago(10m) and extent_id() == _extentId;
// This invokes the function in the update policy (that internally references `MySourceTable`).
MyFunction

事务性设置

更新策略 IsTransactional 设置定义了更新策略是否是事务性策略，并且可影响策略更新的行为，如下所示：

• IsTransactional:false：如果该值设置为默认值“false”，则更新策略不能保证源表和目标表中数据之间的一致性。 如果更新策略失败，则仅将数据引入源表，而不引入目标表。 在这种情况下，引入操作是成功的。
• IsTransactional:true：如果该值设置为“true”，则此设置可以保证源表和目标表中的数据之间的一致性。 如果更新策略失败，则不会将数据引入源表或目标表。 在这种情况下，引入操作是不成功的。

处理故障

策略更新失败时，会根据 IsTransactional 设置是 true 还是 false 进行不同的处理。 更新策略失败的常见原因如下：

• 查询输出架构与目标表不匹配。
• 任何查询错误。

可以使用将 .show ingestion failures 命令与以下命令结合使用来查看策略更新失败：在任何其他情况下，都可以手动重试引入。

.show ingestion failures
| where FailedOn > ago(1hr) and OriginatesFromUpdatePolicy == true

提取、转换和加载的示例

可以使用“更新策略设置”执行提取、转换和加载 (ETL)。

在本示例中，我们将更新策略与简单函数结合使用来执行 ETL。 首先创建两个表：

• 源表 - 包含引入数据的单个字符串类型列。
• 目标表 - 包含所需的架构。 更新策略是在此表上定义的。

1. 让我们创建源表：

   .create table MySourceTable (OriginalRecord:string)
   

2. 接下来创建目标表：

   .create table MyTargetTable (Timestamp:datetime, ThreadId:int, ProcessId:int, TimeSinceStartup:timespan, Message:string)
   

3. 然后创建一个函数来提取数据：

   .create function
    with (docstring = 'Parses raw records into strongly-typed columns', folder = 'UpdatePolicyFunctions')
        ExtractMyLogs()
       {
       MySourceTable
       | parse OriginalRecord with "[" Timestamp:datetime "] [ThreadId:" ThreadId:int "] [ProcessId:" ProcessId:int "] TimeSinceStartup: " TimeSinceStartup:timespan " Message: " Message:string
       | project-away OriginalRecord
   }
   

4. 现在，将更新策略设置为调用创建的函数：

   .alter table MyTargetTable policy update
   @'[{ "IsEnabled": true, "Source": "MySourceTable", "Query": "ExtractMyLogs()", "IsTransactional": true, "PropagateIngestionProperties": false}]'
   

5. 若要在将数据引入目标表后清空源表，请定义源表上的保留策略，使它将 0s 作为其 SoftDeletePeriod。

    .alter-merge table MySourceTable policy retention softdelete = 0s
   

通配符更新策略示例

以下示例在表中 TargetTable创建一个包含单个条目的更新策略。 该策略引用所有匹配模式 SourceTable* 的表作为其源。 与模式（在本地数据库中）匹配的表的任何引入都将根据更新策略查询触发更新策略，并将数据引入到 TargetTable该表。

1. 创建两个源表：

   .create table SourceTable1(Id:long, Value:string)
   

   .create table SourceTable2(Id:long, Value:string)
   

2. 创建目标表：

   .create table TargetTable(Id:long, Value:string, Source:string)
   

3. 创建将用作 Query 更新策略的函数。 该函数使用 $source_table 符号来引用 Source 更新策略。 用于 skipValidation=true 在创建函数期间跳过验证，因为在 $source_table 更新策略执行期间才知道。 更改更新策略时，在下一步中验证该函数。

   .create function with(skipValidation=true) IngestToTarget()
   {
       $source_table 
       | parse Value with "I'm from table " Source
       | project Id, Value, Source
   }
   

4. 创建更新策略。TargetTable 该策略引用所有匹配模式 SourceTable* 的表作为其源。

       .alter table TargetTable policy update
       ```[{ 
               "IsEnabled": true, 
               "Source": "SourceTable*", 
               "SourceIsWildCard" : true,
               "Query": "IngestToTarget()",
               "IsTransactional": true,
               "PropagateIngestionProperties": true
       }]```
   
   

5. 引入源表。 这两个引入都触发更新策略：

   .set-or-append SourceTable1 <| 
       datatable (Id:long, Value:string)
       [
           1, "I'm from table SourceTable1",
           2, "I'm from table SourceTable1"
       ]
   

   .set-or-append SourceTable2 <| 
       datatable (Id:long, Value:string)
       [
           3, "I'm from table SourceTable2",
           4, "I'm from table SourceTable2"
       ]
   

6. 查询 TargetTable：

    TargetTable
   

   Id	价值	源
   1	我来自表 SourceTable1	SourceTable1
   2	我来自表 SourceTable1	SourceTable1
   3	我来自表 SourceTable2	SourceTable2
   4	我来自表 SourceTable2	SourceTable2

相关内容

• 使用表更新策略的常见方案
• 教程：使用表更新策略路由数据