Azure 監視器中的數據收集轉換

  • 發行項

透過 Azure 監視器中的轉換，您可以在將資料傳送至 Log Analytics 工作區之前篩選或修改傳入數據。 本文提供轉換的基本描述及其實作方式。 它提供建立轉換之其他內容的連結。

轉換會在數據擷取管線中的 Azure 監視器中執行，然後數據源傳遞數據，再傳送至目的地。 數據源可能會在傳送數據之前執行自己的篩選，但在傳送至目的地之前，先依賴轉換進行進一步操作。

轉換定義在資料收集規則 （DCR） 中，並使用個別套用至傳入數據中每個專案的 Kusto 查詢語言 （KQL） 語句。 它必須瞭解傳入數據的格式，並在目的地預期的結構中建立輸出。

下圖說明傳入數據的轉換程式，並顯示可能使用的範例查詢。 如需建置轉換查詢的詳細資訊，請參閱 Azure 監視器 中的轉換結構。

顯示內送數據擷取時間轉換的圖表。

為何要使用轉換

下表說明您可以使用轉換達成的不同目標。

類別 詳細資料
拿掉敏感數據 您可能會有一個數據源，其會傳送您不想基於隱私權或相容性原因儲存的資訊。

篩選敏感性資訊。 篩選掉包含敏感性信息的整個數據列或特定數據行。

模糊化敏感性資訊。 以一般字元取代IP位址或電話號碼中的數位等資訊。

傳送至替代數據表。 將敏感性記錄傳送至具有不同角色型訪問控制組態的替代數據表。
使用更多或計算的信息擴充數據 使用轉換將資訊新增至數據，以提供商務內容，或稍後簡化查詢數據。

新增具有詳細信息的數據行。 例如，您可以新增資料行，以識別另一個數據行中的IP位址是內部或外部的。

新增商務特定資訊。 例如，您可以新增數據行，以根據其他數據行中的位置資訊來指出公司部門。
降低數據成本 因為您需支付傳送至 Log Analytics 工作區之任何數據的擷取成本，因此您想要篩選掉不需要的任何數據，以降低成本。

拿掉整個數據列。 例如，您可能有診斷設定可從特定資源收集資源記錄，但不需要它產生的所有記錄專案。 建立轉換，篩選出符合特定準則的記錄。

從每個數據列移除一個數據行。 例如，您的數據可能包含具有備援或最小值的數據行。 建立轉換，以篩選出不需要的數據行。

從數據行剖析重要數據。 您可能會有一個數據表，其中含有特定數據行中埋藏的寶貴數據。 使用轉換將寶貴的數據剖析成新的數據行，並移除原始數據。

將特定數據列傳送至基本記錄。 將數據列中需要基本查詢功能的數據列傳送至基本記錄數據表，以降低擷取成本。
格式化目的地的數據 您可能有一個數據源，其會以不符合目的地數據表結構的格式傳送數據。 使用轉換將數據重新格式化為必要的架構。

支援的資料表

如需可用於轉換的數據表清單，請參閱 支援 Azure 監視器 記錄中轉換的數據表。 您也可以使用 Azure 監視器資料參考 ，其中列出每個數據表的屬性，包括它是否支持轉換。 除了這些數據表之外，也支援任何自定義數據表（_CL後綴）。

建立轉換

根據數據收集方法，有多個方法可以建立轉換。 下表列出建立轉換之不同方法的指引。

資料集合 參考
記錄擷取 API 使用 REST API 將資料傳送至 Azure 監視器記錄 （Azure 入口網站）
使用 REST API 將資料傳送至 Azure 監視器記錄 （Azure Resource Manager 樣本）
具有 Azure 監視器代理程式的虛擬機 將轉換新增至 Azure 監視器記錄
使用容器深入解析的 Kubernetes 叢集 容器深入解析中的數據轉換
Azure 事件中樞 教學課程：從 Azure 事件中樞 擷取事件到 Azure 監視器記錄 （公開預覽）

多個目的地

透過轉換，您可以使用單一 DCR 將數據傳送至 Log Analytics 工作區中的多個目的地。 您可以為每個目的地提供 KQL 查詢，並將每個查詢的結果傳送至其對應的位置。 您可以將不同的數據集傳送至不同的數據表，或使用多個查詢將不同的數據集傳送至相同的數據表。

例如，您可以使用記錄擷取 API 將事件資料傳送至 Azure 監視器。 大部分的事件都應該傳送分析數據表，以便定期查詢，而稽核事件應傳送至針對基本記錄設定的自定義數據表，以降低成本。

若要使用多個目的地，您目前必須手動建立新的 DCR 或 編輯現有的 DCR。 如需使用多個目的地的 DCR 範例，請參閱範例一節。

重要

目前，DCR 中的數據表必須位於相同的Log Analytics工作區中。 若要從單一數據源傳送至多個工作區，請使用多個 DCR 並設定應用程式將數據傳送至每個工作區。

此圖顯示將數據傳送至多個數據表的轉換。

監視轉換

如需監視轉換健康情況和效能的記錄和計量的詳細資訊，請參閱 在 Azure 監視器 中監視 DCR 數據收集並進行疑難解答。 這包括識別 KQL 中發生的任何錯誤，以及用來追蹤其執行持續時間的計量。

轉換的成本

雖然轉換本身不會產生直接成本，但下列案例可能會導致額外費用：

  • 如果轉換增加傳入數據的大小，例如新增匯出數據行，您將需支付額外數據的標準擷取費率。
  • 如果轉換減少擷取的數據超過 50%，您將需支付超過 50% 的篩選數據量的費用。

若要計算轉換所產生的數據處理費用，請使用下列公式：
[依轉換篩選出的 GB] - （[以管線擷取的 GB 數據] / 2）。 下表顯示範例。

管線擷取的數據 轉換捨棄的數據 Log Analytics 工作區所擷取的數據 數據處理費用 擷取費用
20 GB 12 GB 8 GB 2 GB 1 8 GB
20 GB 8 GB 12 GB 0 GB 12 GB

1 此費用不包括Log Analytics工作區內嵌的數據費用。

若要避免這項費用，您應該先使用替代方法篩選內嵌的數據，再套用轉換。 如此一來，您就可以減少轉換所處理的數據量，因此，將任何額外的成本降到最低。

如需擷取和保留 Azure 監視器中記錄數據的目前費用，請參閱 Azure 監視器定價 。

重要

如果已啟用 Log Analytics 工作區的 Azure Sentinel，則不論轉換篩選會篩選多少數據，都沒有任何篩選費用。

範例

下列 Resource Manager 樣本顯示具有不同模式的範例 DCR。 您可以使用這些範本作為起點，為您自己的案例建立具有轉換的 DCR。

單一目的地

下列範例是 Azure 監視器代理程式的 DCR，可將數據傳送至 Syslog 數據表。 在此範例中，轉換會篩選訊息中具有 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": "eastus", 
            "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 中將數據傳送至 Syslog 和 SecurityEvent 數據表的數據 DCR。 此 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": "eastus", 
            "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，可將數據傳送至 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": "eastus", 
            "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" 
                    } 
                ] 
            } 
        }
    ]
}

下一步