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

适用于 Python 的 Azure Monitor 查询客户端库 - 版本 1.2.0

项目
05/10/2023

Azure Monitor 查询客户端库用于对 Azure Monitor 的两个数据平台执行只读查询：

日志 - 从受监视的资源收集和组织日志和性能数据。来自不同源的数据（例如来自 Azure 服务的平台日志、来自虚拟机代理的日志和性能数据，以及应用的使用情况和性能数据）可以合并到单个 Azure Log Analytics 工作区中。可以使用 Kusto 查询语言一起分析各种数据类型。
指标 - 将受监视资源中的数字数据收集到时序数据库中。指标是定期收集的数值，用于描述系统在某一特定时间的某些情况。指标是轻量级的，能够支持准实时方案，因此可用于发出警报和快速检测问题。

资源：

入门

先决条件

Python 3.7 或更高版本
Azure 订阅
TokenCredential 实现，例如 Azure 标识库凭据类型。
若要查询日志，需要一个 Azure Log Analytics 工作区。
若要查询指标，需要 (存储帐户、密钥保管库、Cosmos DB 等) 的任何类型的 Azure 资源。

安装包

使用 pip 安装适用于 Python 的 Azure Monitor 查询客户端库：

pip install azure-monitor-query

创建客户端

查询日志或指标需要经过身份验证的客户端。该库包括客户端的同步和异步形式。若要进行身份验证，请创建令牌凭据的实例。创建 LogsQueryClient 或 MetricsQueryClient时使用该实例。以下示例使用 DefaultAzureCredentialazure-identity 包中的。

同步客户端

请考虑以下示例，该示例为日志和指标查询创建同步客户端：

from azure.identity import DefaultAzureCredential
from azure.monitor.query import LogsQueryClient, MetricsQueryClient

credential = DefaultAzureCredential()
logs_client = LogsQueryClient(credential)
metrics_client = MetricsQueryClient(credential)

异步客户端

查询客户端 API 的异步形式位于后缀命名空间中 .aio。例如：

from azure.identity.aio import DefaultAzureCredential
from azure.monitor.query.aio import LogsQueryClient, MetricsQueryClient

credential = DefaultAzureCredential()
async_logs_client = LogsQueryClient(credential)
async_metrics_client = MetricsQueryClient(credential)

为非公有 Azure 云配置客户端

默认情况下， LogsQueryClient 和 MetricsQueryClient 配置为连接到公共 Azure 云。可以通过传入正确的 endpoint 参数将这些云配置为连接到非公有 Azure 云：例如：

logs_client = LogsQueryClient(credential, endpoint="https://api.loganalytics.azure.cn/v1")
metrics_client = MetricsQueryClient(credential, endpoint="https://management.chinacloudapi.cn")

注意：目前， MetricsQueryClient 使用 Azure 资源管理器 (ARM) 终结点来查询指标，因此，使用此客户端时需要云的相应管理终结点。将来可能会发生变化。

执行查询

有关日志和指标查询的示例，请参阅示例部分。

关键概念

日志查询速率限制和限制

当请求速率过高时，Log Analytics 服务会应用限制。限制（如返回的最大行数）也应用于 Kusto 查询。有关详细信息，请参阅查询 API。

如果要执行批处理日志查询，则受限制的请求将返回对象 LogsQueryError 。该对象的 code 值为 ThrottledError。

指标数据结构

每组指标值都是具有以下特征的时序：

值的收集时间
与值关联的资源
充当指标类别的命名空间
指标名称
值本身
某些指标可能具有多个维度，如多维指标中所述。自定义指标最多可以包含 10 个维度。

日志查询

此示例演示如何查询 Log Analytics 工作区。若要处理响应，并用表格形式查看响应，请使用 pandas 库。如果选择不使用 pandas，请参阅示例。

指定时间跨度

参数 timespan 指定查询数据的持续时间。此值可以为下列值之一：

a timedelta
a timedelta 和开始日期/时间
开始日期/结束日期/时间

例如：

import os
import pandas as pd
from datetime import datetime, timezone
from azure.monitor.query import LogsQueryClient, LogsQueryStatus
from azure.identity import DefaultAzureCredential
from azure.core.exceptions import HttpResponseError

credential = DefaultAzureCredential()
client = LogsQueryClient(credential)

query = """AppRequests | take 5"""

start_time=datetime(2021, 7, 2, tzinfo=timezone.utc)
end_time=datetime(2021, 7, 4, tzinfo=timezone.utc)

try:
    response = client.query_workspace(
        workspace_id=os.environ['LOG_WORKSPACE_ID'],
        query=query,
        timespan=(start_time, end_time)
        )
    if response.status == LogsQueryStatus.PARTIAL:
        error = response.partial_error
        data = response.partial_data
        print(error)
    elif response.status == LogsQueryStatus.SUCCESS:
        data = response.tables
    for table in data:
        df = pd.DataFrame(data=table.rows, columns=table.columns)
        print(df)
except HttpResponseError as err:
    print("something fatal happened")
    print(err)

处理日志查询响应

API query_workspace 返回 LogsQueryResult 或 LogsQueryPartialResult 对象。 API batch_query 返回可能包含 LogsQueryResult、 LogsQueryPartialResult和 LogsQueryError 对象的列表。下面是响应的层次结构：

LogsQueryResult
|---statistics
|---visualization
|---tables (list of `LogsTable` objects)
    |---name
    |---rows
    |---columns
    |---columns_types

LogsQueryPartialResult
|---statistics
|---visualization
|---partial_error (a `LogsQueryError` object)
    |---code
    |---message
    |---details
    |---status
|---partial_data (list of `LogsTable` objects)
    |---name
    |---rows
    |---columns
    |---columns_types

LogsQueryResult为方便起见，直接循环访问表。例如，若要使用表处理日志查询响应并使用 pandas 显示它，请执行以下操作：

response = client.query(...)
for table in response:
    df = pd.DataFrame(table.rows, columns=[col.name for col in table.columns])

可在此处找到完整示例。

以类似的方式处理批处理日志查询响应：

for result in response:
    if result.status == LogsQueryStatus.SUCCESS:
        for table in result:
            df = pd.DataFrame(table.rows, columns=table.columns)
            print(df)

可在此处找到完整示例。

批处理日志查询

以下示例演示如何使用批处理查询 API 同时发送多个查询。查询可以表示为对象列表 LogsBatchQuery 或字典。此示例使用前一种方法。

import os
from datetime import timedelta, datetime, timezone
import pandas as pd
from azure.monitor.query import LogsQueryClient, LogsBatchQuery, LogsQueryStatus
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = LogsQueryClient(credential)
requests = [
    LogsBatchQuery(
        query="AzureActivity | summarize count()",
        timespan=timedelta(hours=1),
        workspace_id=os.environ['LOG_WORKSPACE_ID']
    ),
    LogsBatchQuery(
        query= """bad query""",
        timespan=timedelta(days=1),
        workspace_id=os.environ['LOG_WORKSPACE_ID']
    ),
    LogsBatchQuery(
        query= """let Weight = 92233720368547758;
        range x from 1 to 3 step 1
        | summarize percentilesw(x, Weight * 100, 50)""",
        workspace_id=os.environ['LOG_WORKSPACE_ID'],
        timespan=(datetime(2021, 6, 2, tzinfo=timezone.utc), datetime(2021, 6, 5, tzinfo=timezone.utc)), # (start, end)
        include_statistics=True
    ),
]
results = client.query_batch(requests)

for res in results:
    if res.status == LogsQueryStatus.FAILURE:
        # this will be a LogsQueryError
        print(res.message)
    elif res.status == LogsQueryStatus.PARTIAL:
        ## this will be a LogsQueryPartialResult
        print(res.partial_error)
        for table in res.partial_data:
            df = pd.DataFrame(table.rows, columns=table.columns)
            print(df)
    elif res.status == LogsQueryStatus.SUCCESS:
        ## this will be a LogsQueryResult
        table = res.tables[0]
        df = pd.DataFrame(table.rows, columns=table.columns)
        print(df)

资源日志查询

以下示例演示如何直接从 Azure 资源查询日志，而无需使用 Log Analytics 工作区。在这里， query_resource 使用方法而不是 query_workspace，而不是工作区 ID，而是传入 (例如 /subscriptions/{subscription-id}/resourceGroups/{resource-group-name}/providers/{resource-provider}/{resource-type}/{resource-name}) 。

import os
import pandas as pd
from datetime import timedelta
from azure.monitor.query import LogsQueryClient, LogsQueryStatus
from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential

credential  = DefaultAzureCredential()
client = LogsQueryClient(credential)

query = """AzureActivity | take 5"""

try:
    response = client.query_resource(os.environ['LOGS_RESOURCE_ID'], query, timespan=timedelta(days=1))
    if response.status == LogsQueryStatus.PARTIAL:
        error = response.partial_error
        data = response.partial_data
        print(error)
    elif response.status == LogsQueryStatus.SUCCESS:
        data = response.tables
    for table in data:
        df = pd.DataFrame(data=table.rows, columns=table.columns)
        print(df)
except HttpResponseError as err:
    print("something fatal happened")
    print(err)

高级日志查询方案

设置日志查询超时

以下示例演示如何设置服务器超时（以秒为单位）。如果查询花费的时间超过上述超时时间，则会引发网关超时。默认值为 180 秒，最多可以设置为 10 分钟 (600 秒) 。

import os
from azure.monitor.query import LogsQueryClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = LogsQueryClient(credential)

response = client.query_workspace(
    os.environ['LOG_WORKSPACE_ID'],
    "range x from 1 to 10000000000 step 1 | count",
    timespan=timedelta(days=1),
    server_timeout=600 # sets the timeout to 10 minutes
    )

查询多个工作区

可以在多个 Log Analytics 工作区中执行相同的日志查询。除了 Kusto 查询之外，还需要以下参数：

workspace_id - 第一个 (主) 工作区 ID。
additional_workspaces - 工作区列表，不包括参数中 workspace_id 提供的工作区。参数的列表项可能包含以下标识符格式：
- 限定的工作区名称
- 工作区 ID
- Azure 资源 ID

例如，以下查询在三个工作区中执行：

client.query_workspace(
    <workspace_id>,
    query,
    timespan=timedelta(days=1),
    additional_workspaces=['<workspace 2>', '<workspace 3>']
    )

可在此处找到完整示例。

包括统计信息

若要获取日志查询执行统计信息，例如 CPU 和内存消耗：

将 include_statistics 参数设置为 True。
statistics访问对象内的 LogsQueryResult 字段。

以下示例输出查询执行时间：

query = "AzureActivity | top 10 by TimeGenerated"
result = client.query_workspace(
    <workspace_id>,
    query,
    timespan=timedelta(days=1),
    include_statistics=True
    )

execution_time = result.statistics.get("query", {}).get("executionTime")
print(f"Query execution time: {execution_time}")

字段 statistics 是 dict 对应于原始 JSON 响应的，其结构可能因查询而异。统计信息位于属性中 query 。例如：

{
  "query": {
    "executionTime": 0.0156478,
    "resourceUsage": {...},
    "inputDatasetStatistics": {...},
    "datasetStatistics": [{...}]
  }
}

包括可视化效果

若要使用 render 运算符获取日志查询的可视化数据，请执行以下操作：

将 include_visualization 属性设置为 True。
visualization访问对象内的 LogsQueryResult 字段。

例如：

query = (
    "StormEvents"
    "| summarize event_count = count() by State"
    "| where event_count > 10"
    "| project State, event_count"
    "| render columnchart"
)
result = client.query_workspace(
    <workspace_id>,
    query,
    timespan=timedelta(days=1),
    include_visualization=True
    )

print(f"Visualization result: {result.visualization}")

字段 visualization 是 dict 对应于原始 JSON 响应的，其结构可能因查询而异。例如：

{
  "visualization": "columnchart",
  "title": "the chart title",
  "accumulate": False,
  "isQuerySorted": False,
  "kind": None,
  "legend": None,
  "series": None,
  "yMin": "NaN",
  "yMax": "NaN",
  "xAxis": None,
  "xColumn": None,
  "xTitle": "x axis title",
  "yAxis": None,
  "yColumns": None,
  "ySplit": None,
  "yTitle": None,
  "anomalyColumns": None
}

指标查询

以下示例获取事件网格订阅的指标。资源 URI 是事件网格主题的资源 URI。

资源 URI 必须是要查询其指标的资源的 URI。它通常采用格式 /subscriptions/<id>/resourceGroups/<rg-name>/providers/<source>/topics/<resource-name>。

查找资源 URI：

导航到Azure 门户中的资源页。
在 “概述 ”边栏选项卡中，选择“ JSON 视图” 链接。
在生成的 JSON 中，复制属性的值 id 。

注意：指标按发送metric_names的顺序返回。

import os
from datetime import timedelta, datetime
from azure.monitor.query import MetricsQueryClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = MetricsQueryClient(credential)
start_time = datetime(2021, 5, 25)
duration = timedelta(days=1)
metrics_uri = os.environ['METRICS_RESOURCE_URI']
response = client.query_resource(
    metrics_uri,
    metric_names=["PublishSuccessCount"],
    timespan=(start_time, duration)
    )

for metric in response.metrics:
    print(metric.name)
    for time_series_element in metric.timeseries:
        for metric_value in time_series_element.data:
            print(metric_value.time_stamp)

处理指标查询响应

指标查询 API 返回对象 MetricsQueryResult 。对象MetricsQueryResult包含一些属性，例如类型化对象、granularity、 namespace和 timespan的列表Metric。 Metric可以使用参数访问metrics对象列表。此列表中的每个对象都包含 Metric 一个 TimeSeriesElement 对象列表。每个 TimeSeriesElement 对象都包含 data 和 metadata_values 属性。在可视形式中，响应的对象层次结构类似于以下结构：

MetricsQueryResult
|---granularity
|---timespan
|---cost
|---namespace
|---resource_region
|---metrics (list of `Metric` objects)
    |---id
    |---type
    |---name
    |---unit
    |---timeseries (list of `TimeSeriesElement` objects)
        |---metadata_values
        |---data (list of data points represented by `MetricValue` objects)

处理响应的示例

import os
from azure.monitor.query import MetricsQueryClient, MetricAggregationType
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = MetricsQueryClient(credential)

metrics_uri = os.environ['METRICS_RESOURCE_URI']
response = client.query_resource(
    metrics_uri,
    metric_names=["MatchedEventCount"],
    aggregations=[MetricAggregationType.COUNT]
    )

for metric in response.metrics:
    print(metric.name)
    for time_series_element in metric.timeseries:
        for metric_value in time_series_element.data:
            if metric_value.count != 0:
                print(
                    "There are {} matched events at {}".format(
                        metric_value.count,
                        metric_value.time_stamp
                    )
                )

疑难解答

有关如何诊断各种故障方案的详细信息，请参阅故障排除指南。

后续步骤

若要了解有关 Azure Monitor 的详细信息，请参阅 Azure Monitor 服务文档。

示例

以下代码示例演示 Azure Monitor 查询客户端库的常见方案。

日志查询示例

指标查询示例

贡献

本项目欢迎贡献和建议。大多数贡献要求你同意贡献者许可协议 (CLA)，并声明你有权（并且确实有权）授予我们使用你的贡献的权利。有关详细信息，请访问 cla.microsoft.com。

提交拉取请求时，CLA 机器人将自动确定你是否需要提供 CLA，并相应地修饰 PR（例如标签、注释）。直接按机器人提供的说明操作。只需使用 CLA 在所有存储库中执行此操作一次。

此项目采用了 Microsoft 开放源代码行为准则。有关详细信息，请参阅行为准则常见问题解答，或如果有任何其他问题或意见，请与联系。

通过

适用于 Python 的 Azure Monitor 查询客户端库 - 版本 1.2.0

入门

先决条件

安装包

创建客户端

同步客户端

异步客户端

为非公有 Azure 云配置客户端

执行查询

关键概念

日志查询速率限制和限制

指标数据结构

示例

日志查询

指定时间跨度

处理日志查询响应

批处理日志查询

资源日志查询

高级日志查询方案

设置日志查询超时

查询多个工作区

包括统计信息

包括可视化效果

指标查询

处理指标查询响应

处理响应的示例

疑难解答

后续步骤

示例

日志查询示例

指标查询示例

贡献

其他资源