Azure Monitor 中的数据收集转换

项目
04/16/2024

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

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

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

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

为何使用转换

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

类别	详细信息
删除敏感数据	你可能有一个数据源，用于发送不希望出于隐私或合规性原因而存储的信息。筛选敏感信息。筛选出整行或筛选出包含敏感信息的特定列。混淆敏感信息。将 IP 地址或电话号码中的数字等信息替换为通用字符。发送到备用表。将敏感记录发送到具有不同基于角色的访问控制配置的备用表。
使用更多或计算的信息扩充数据	使用转换将信息添加到数据，从而提供业务上下文或稍后简化查询数据。添加包含更多信息的列。例如，你可能会添加一个列，用于确定另一列中的 IP 地址是内部地址还是外部地址。添加特定于业务的信息。例如，你可能会添加一个列，指示基于其他列中的位置信息的公司部门。
降低数据成本	由于需要为任何发送到 Log Analytics 工作区的数据支付引入费用，因此需要筛选出所有不需要的数据来降低成本。删除整行。例如，你可能有一个诊断设置来从特定资源收集资源日志，但不需要它生成的所有日志条目。创建一个筛选掉符合特定条件的记录的转换。从每行中删除列。例如，数据可能包含具有冗余数据或最小值的列。创建一个筛选出不需要的列的转换。分析列中的重要数据。你可能有一个表，其中包含隐藏在特定列中的有价值数据。使用转换将有价值的数据解析为新列并删除原始数据。将某些行发送到基本日志。将数据中需要基本查询功能的行发送到基本日志表，以降低引入成本。
设置目标数据的格式	你可能有数据源，该数据源以与目标表的结构不匹配的格式发送数据。使用转换将数据重格式化为所需架构。

受支持的表

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

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

创建转换

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

数据收集	参考
日志引入 API	使用 REST API 将数据发送到 Azure Monitor 日志（Azure 门户）使用 REST API（Azure 资源管理器模板）将数据发送到 Azure Monitor 日志
具有 Azure Monitor 代理的虚拟机	向 Azure Monitor 日志添加转换
具有容器见解的 Kubernetes 群集	容器见解中的数据转换

多个目标

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

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

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

重要

目前，DCR 中的表必须位于同一 Log Analytics 工作区中。若要从单个数据源发送到多个工作区，请使用多个 DCR 并将应用程序配置为将数据发送到每个工作区。

监视转换

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

转换的成本

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

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

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

管道引入的数据	转换删除的数据	Log Analytics 工作区引入的数据	数据处理费用	引入费用
20 GB	12 GB	8 GB	2 GB ¹	8 GB
20 GB	8 GB	12 GB	0 GB	12 GB

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

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

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

重要

如果为 Log Analytics 工作区启用了 Azure Sentinel，则无论转换筛选的数据量如何，都不会产生筛选引入费用。

示例

以下资源管理器模板显示具有不同模式的示例 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" 
                    } 
                ] 
            } 
        }
    ]
}