Fabric Capacity指标数据获取：REST API查询及Python实现方案问询

一、REST API 方式 #

有官方REST API可以直接拉取你需要的容量使用数据，不用绕内置应用的数据集：

• Capacity Get Refreshables：能获取容量刷新作业的相关指标，包含CPU使用率、峰值这类维度数据
• Capacity Metrics Get：专门用来提取容量的实时和历史使用指标，覆盖CPU、内存、IO等资源的时间序列数据，支持按时间范围过滤，完全匹配你要的随时间变化的峰值需求
• 权限方面：你有管理员权限，只要请求里带上有效的Azure AD OAuth2令牌就能访问这些接口

二、Python替代方案（如果API用不了的情况） #

如果因为环境限制没法用REST API，这些Python方法也能搞定：

1. 用Power BI Python SDK（azure-mgmt-powerbiembedded） #

• 先装依赖：pip install azure-mgmt-powerbiembedded azure-identity
• 核心思路：通过Azure Identity拿身份令牌，用SDK封装的接口直接拉结构化数据
• 代码示例：

from azure.identity import DefaultAzureCredential
from azure.mgmt.powerbiembedded import PowerBIEmbeddedManagementClient

credential = DefaultAzureCredential()
client = PowerBIEmbeddedManagementClient(credential, "<你的订阅ID>")

# 获取指定容量的指标数据，这里过滤了CPU百分比、1小时粒度、指定时间范围
metrics = client.capacities.get_metrics(
    resource_group_name="<你的资源组名>",
    capacity_name="<你的容量名>",
    filter="name.value eq 'CpuPercentage' and timeGrain eq 'PT1H' and startTime eq '2024-01-01T00:00:00Z' and endTime eq '2024-01-02T00:00:00Z'"
)

# 解析数据，这里可以改成保存到文件或对象存储
for metric in metrics.value:
    for time_series in metric.timeseries:
        for data_point in time_series.data:
            print(f"时间: {data_point.time_stamp}, CPU使用率: {data_point.average}")

2. 直接读取内置的Fabric Capacity Usage metrics数据集 #

• 用powerbiclient或pyodbc连接数据集：
  • 装依赖：pip install powerbiclient pyodbc
  • 核心思路：用管理员身份认证后，连接到内置应用的数据集，写DAX查询提取CPU和峰值数据
• 代码示例：

from powerbiclient.authentication import InteractiveLoginAuthentication
from powerbiclient.dataset import DatasetClient
import pandas as pd

# 管理员身份交互式登录
auth = InteractiveLoginAuthentication()

# 从Power BI界面获取内置应用的数据集ID和工作区ID
dataset_id = "<Fabric容量使用数据集ID>"
group_id = "<内置应用所在工作区ID>"

# 写DAX查询，提取时间戳、容量名和CPU峰值
dax_query = """
EVALUATE
SUMMARIZECOLUMNS(
    'Capacity Usage'[TimeStamp],
    'Capacity Usage'[CapacityName],
    "CPU_Percentage", MAX('Capacity Usage'[CPUPercentage])
)
ORDER BY 'Capacity Usage'[TimeStamp] DESC
"""

# 执行查询并转成DataFrame
dataset_client = DatasetClient(auth=auth, group_id=group_id, dataset_id=dataset_id)
result = dataset_client.execute_dax_query(dax_query)
df = pd.DataFrame(result['results'][0]['tables'][0]['rows'])

# 先保存到本地CSV，后续再上传到对象存储
df.to_csv("cpu_usage_data.csv", index=False)

3. 定期上传到对象存储 #

拿到数据后，用对应云服务商的Python SDK就能定期上传，比如Azure Blob Storage：

from azure.storage.blob import BlobServiceClient
import pandas as pd

# 替换成你的存储连接字符串和容器名
blob_service_client = BlobServiceClient.from_connection_string("<你的存储连接字符串>")
container_client = blob_service_client.get_container_client("<你的容器名>")

# 按日期命名文件，避免覆盖
file_name = f"cpu_usage_{pd.Timestamp.now().strftime('%Y%m%d')}.csv"
with open("cpu_usage_data.csv", "rb") as data:
    container_client.upload_blob(name=file_name, data=data, overwrite=True)

内容的提问来源于stack exchange，提问作者Ameya Bhave