Azure Monitor 中的数据收集转换

通过 Azure Monitor 中的转换，可在将传入数据发送到 Log Analytics 工作区之前对其进行筛选或修改。 本文提供转换的基本说明及其实现方式。 它提供指向用于创建转换的其他内容的链接。

在数据源传递数据之后且在将数据发送到目标之前，在数据引入管道的 Azure Monitor 中执行转换。 数据源可以在发送数据之前执行自己的筛选，但将其发送到目标之前，需依赖转换来进一步操作。

转换在数据收集规则 (DCR) 中定义，并使用单独应用于传入数据中的每个条目的 Kusto 查询语言 (KQL) 语句。 它必须了解传入数据的格式，并在目标所需的结构中创建输出。

下图演示了传入数据的转换过程，并显示了可能使用的示例查询。 有关生成转换查询的详细信息，请参阅Azure Monitor 中的转换结构。

为何使用转换

下表描述了可使用转换实现的不同目标。

受支持的表

有关可用于转换的表列表，请参阅支持 Azure Monitor 日志中转换的表。 还可以使用Azure Monitor 数据引用，其中列出了每个表的属性，包括它是否支持转换。 除了这些表，还支持任何自定义表（_CL后缀）。

• Azure Monitor 日志中支持转换的表中列出的任何 Azure 表。 还可以使用Azure Monitor 数据引用，其中列出了每个表的属性，包括它是否支持转换。
• 为 Azure Monitor 代理创建的任意自定义表。 （MMA 自定义表无法使用转换）

创建转换

有多种方法可以创建转换，具体取决于数据收集方法。 下表列出了用于创建转换的不同方法的指导。

多个目标

通过转换，可以使用单个 DCR 将数据发送到 Log Analytics 工作区中的多个目标。 可为每个目标提供一个 KQL 查询，并将每个查询的结果发送到其相应位置。 可以将不同数据集发送到不同的表，或使用多个查询将不同的数据集发送到同一个表。

例如，可以使用日志引入 API 将事件数据发送到 Azure Monitor。 大多数事件应发送到分析表（可在其中定期查询），而审核事件应发送到为基本日志配置的自定义表，以降低成本。

若要使用多个目标，当前必须手动创建新的 DCR 或编辑现有 DCR。 有关使用多个目标的 DCR 的示例，请参阅示例部分。

监视转换

请参阅在 Azure Monitor 中监视 DCR 数据收集和排查相关问题，详细了解用于监视转换运行状况和性能的日志和指标。 这包括识别 KQL 中发生的任何错误以及用于跟踪其运行持续时间的指标。

转换的成本

虽然转换本身不会产生直接成本，但以下情况可能会导致额外费用：

• 如果转换增加了传入数据的大小（例如通过添加计算列），则需要为额外数据支付标准引入费。
• 如果转换将传入数据减少了 50% 以上，则需要为超过 50% 的筛选数据量付费。

若要计算转换产生的数据处理费用，请使用以下公式：
[通过转换筛选掉的 GB] -([引入的总 GB] / 2)。 下表显示了示例。

¹ 此费用不包括 Log Analytics 工作区引入的数据的费用。

为了避免产生此费用，应在应用转换之前使用替代方法筛选引入的数据。 通过这样做，可以减少转换处理的数据量，从而最大限度地减少任何额外的成本。

请参阅 Azure Monitor 定价，了解当前在 Azure Monitor 中引入和保留日志数据的费用。

示例

以下资源管理器模板显示具有不同模式的示例 DCR。 你可以使用这些模板作为起点，为你自己的场景创建带有转换的 DCR。

单个目标

以下示例是用于向 Syslog 表发送数据的 Azure Monitor 代理的 DCR。 在此示例中，转换会筛选消息中出现“error”的记录的数据。

{ 
    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "resources" : [
        {
            "type": "Microsoft.Insights/dataCollectionRules", 
            "name": "singleDestinationDCR", 
            "apiVersion": "2021-09-01-preview", 
            "location": "chinanorth2", 
            "properties": { 
              "dataSources": { 
                "syslog": [ 
                  { 
                    "name": "sysLogsDataSource", 
                    "streams": [ 
                      "Microsoft-Syslog" 
                    ], 
                    "facilityNames": [ 
                      "auth",
                      "authpriv",
                      "cron",
                      "daemon",
                      "mark",
                      "kern",
                      "mail",
                      "news",
                      "syslog",
                      "user",
                      "uucp"
                    ], 
                    "logLevels": [ 
                      "Debug", 
                      "Critical", 
                      "Emergency" 
                    ] 
                  } 
                ] 
              }, 
              "destinations": { 
                "logAnalytics": [ 
                  { 
                    "workspaceResourceId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers/Microsoft.OperationalInsights/workspaces/my-workspace", 
                    "name": "centralWorkspace" 
                  } 
                ] 
              }, 
              "dataFlows": [ 
                { 
                  "streams": [ 
                    "Microsoft-Syslog" 
                  ], 
                  "transformKql": "source | where message has 'error'", 
                  "destinations": [ 
                    "centralWorkspace" 
                  ] 
                } 
              ] 
            }
        }
    ]
} 

多个 Azure 表

以下示例是日志引入 API 中数据的 DCR，该 API 会将数据发送到 Syslog 和 SecurityEvent 表。 此 DCR 要求每个表都有单独的 dataFlow，并且每个表具有不同的 transformKql 和 OutputStream。 在此示例中，所有传入数据都发送到 Syslog 表，而恶意数据也发送到 SecurityEvent 表。 如果不想复制这两个表中的恶意数据，则可以将 where 语句添加到第一个查询以删除这些记录。

{ 
    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "resources" : [
        { 
            "type": "Microsoft.Insights/dataCollectionRules", 
            "name": "multiDestinationDCR", 
            "location": "chinanorth2", 
            "apiVersion": "2021-09-01-preview", 
            "properties": { 
                "dataCollectionEndpointId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers//Microsoft.Insights/dataCollectionEndpoints/my-dce",
                "streamDeclarations": { 
                    "Custom-MyTableRawData": { 
                        "columns": [ 
                            { 
                                "name": "Time", 
                                "type": "datetime" 
                            }, 
                            { 
                                "name": "Computer", 
                                "type": "string" 
                            }, 
                            { 
                                "name": "AdditionalContext", 
                                "type": "string" 
                            } 
                        ] 
                    } 
                }, 
                "destinations": { 
                    "logAnalytics": [ 
                        { 
                            "workspaceResourceId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers/Microsoft.OperationalInsights/workspaces/my-workspace", 
                            "name": "clv2ws1" 
                        }, 
                    ] 
                }, 
                "dataFlows": [ 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | project TimeGenerated = Time, Computer, Message = AdditionalContext", 
                        "outputStream": "Microsoft-Syslog" 
                    }, 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | where (AdditionalContext has 'malicious traffic!' | project TimeGenerated = Time, Computer, Subject = AdditionalContext", 
                        "outputStream": "Microsoft-SecurityEvent" 
                    } 
                ] 
            } 
        }
    ]
}

Azure 表和自定义表组合

以下示例是日志引入 API 中数据的 DCR，该 API 会将数据发送到 Syslog 表和具有不同格式数据的自定义表。 此 DCR 要求每个表都有单独的 dataFlow，并且每个表具有不同的 transformKql 和 OutputStream。 使用自定义表时，请务必确保目标（自定义表）的架构包含与要发送的记录架构匹配的自定义列（如何添加或删除自定义列）。 例如，如果记录有一个名为 SyslogMessage 的字段，但目标自定义表只有 TimeGenerated 和 RawData，则将在自定义表中收到一个事件，其中只填充了 TimeGenerated 字段，RawData 字段将为空。 将删除 SyslogMessage 字段，因为目标表的架构不包含名为 SyslogMessage 的字符串字段。

{ 
    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "resources" : [
        { 
            "type": "Microsoft.Insights/dataCollectionRules", 
            "name": "multiDestinationDCR", 
            "location": "chinanorth2", 
            "apiVersion": "2021-09-01-preview", 
            "properties": { 
                "dataCollectionEndpointId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers//Microsoft.Insights/dataCollectionEndpoints/my-dce",
                "streamDeclarations": { 
                    "Custom-MyTableRawData": { 
                        "columns": [ 
                            { 
                                "name": "Time", 
                                "type": "datetime" 
                            }, 
                            { 
                                "name": "Computer", 
                                "type": "string" 
                            }, 
                            { 
                                "name": "AdditionalContext", 
                                "type": "string" 
                            } 
                        ] 
                    } 
                }, 
                "destinations": { 
                    "logAnalytics": [ 
                        { 
                            "workspaceResourceId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers/Microsoft.OperationalInsights/workspaces/my-workspace", 
                            "name": "clv2ws1" 
                        }, 
                    ] 
                }, 
                "dataFlows": [ 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | project TimeGenerated = Time, Computer, SyslogMessage = AdditionalContext", 
                        "outputStream": "Microsoft-Syslog" 
                    }, 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | extend jsonContext = parse_json(AdditionalContext) | project TimeGenerated = Time, Computer, AdditionalContext = jsonContext, ExtendedColumn=tostring(jsonContext.CounterName)", 
                        "outputStream": "Custom-MyTable_CL" 
                    } 
                ] 
            } 
        }
    ]
}

后续步骤

• 阅读有关数据收集规则 (DCR) 的详细信息。
• 创建适用于未使用 DCR 收集的数据的工作区转换 DCR。