如何获取Power BI工作区所有数据集刷新历史并构建数据湖存储管道

核心思路 #

要批量获取工作区内所有数据集的刷新历史，需要分两步执行：先获取工作区内所有数据集的ID列表，再逐个调用单数据集刷新历史接口，最后将整合后的结果同步到Data Lake。

步骤1：获取目标工作区内的所有数据集列表 #

使用Power BI REST API的工作区数据集查询接口，拿到该工作区内所有数据集的基础信息（核心是datasetId）：

GET https://api.powerbi.com/v1.0/myorg/groups/{groupId}/datasets

• groupId：目标工作区的ID，可在Power BI工作区的「设置」页面中找到。
• 身份验证：推荐用**服务主体（Service Principal）**认证，需提前在Power BI Admin中心将服务主体添加到目标工作区，并分配「成员」或「管理员」权限。

步骤2：遍历数据集，批量拉取刷新历史 #

对步骤1返回的每个datasetId，调用单数据集刷新历史接口：

GET https://api.powerbi.com/v1.0/myorg/datasets/{datasetId}/refreshes?$top=100

• $top参数：控制返回的历史记录数量，默认返回最近20条，最大可设为100。
• 建议在结果中关联数据集名称、工作区名称等信息，方便后续的数据分析与溯源。

步骤3：构建数据管道同步至Data Lake #

以下提供两种主流的管道实现方案：

方案A：Azure Data Factory（ADF）企业级管道

1. Web活动获取数据集列表

   • 配置Web活动的URL为步骤1的API地址，认证方式选「Service Principal」，填入租户ID、客户端ID、客户端密钥。
   • 解析返回的JSON响应，提取value数组（所有数据集的集合）。

2. For Each活动遍历数据集

   • 将For Each活动的输入设为步骤1返回的数据集数组，开启「Sequential」模式避免API请求超限。
   • 在For Each内部添加Web活动，调用步骤2的刷新历史API，用当前迭代的datasetId填充URL参数。

3. Copy活动写入Data Lake

   • 将Web活动返回的刷新历史数据作为输入，配置Copy活动的源为「JSON」，接收器为「Azure Data Lake Storage Gen2」。
   • 可设置按日期分区存储（如/refresh-history/{yyyy}/{MM}/{dd}/dataset-{datasetId}.json），便于后续数据管理。

方案B：Python脚本轻量管道

1. 安装依赖库

   pip install requests azure-storage-file-datalake
   

2. 实现API调用与数据同步

   import requests
   from azure.storage.filedatalake import DataLakeServiceClient
   import json
   from datetime import datetime
   
   # Power BI认证信息
   TENANT_ID = "your-tenant-id"
   CLIENT_ID = "your-client-id"
   CLIENT_SECRET = "your-client-secret"
   GROUP_ID = "your-workspace-group-id"
   
   # Data Lake信息
   ADLS_ACCOUNT_NAME = "your-adls-account-name"
   ADLS_CONTAINER_NAME = "your-container-name"
   
   # 获取Power BI访问令牌
   def get_pbi_token():
       url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
       payload = {
           "grant_type": "client_credentials",
           "client_id": CLIENT_ID,
           "client_secret": CLIENT_SECRET,
           "scope": "https://analysis.windows.net/powerbi/api/.default"
       }
       response = requests.post(url, data=payload)
       return response.json()["access_token"]
   
   # 获取工作区所有数据集
   def get_datasets(token):
       url = f"https://api.powerbi.com/v1.0/myorg/groups/{GROUP_ID}/datasets"
       headers = {"Authorization": f"Bearer {token}"}
       response = requests.get(url, headers=headers)
       return response.json()["value"]
   
   # 获取单个数据集刷新历史
   def get_refresh_history(token, dataset_id):
       url = f"https://api.powerbi.com/v1.0/myorg/datasets/{dataset_id}/refreshes?$top=100"
       headers = {"Authorization": f"Bearer {token}"}
       response = requests.get(url, headers=headers)
       return response.json()["value"]
   
   # 上传数据到Data Lake
   def upload_to_adls(data, filename):
       service_client = DataLakeServiceClient(
           account_url=f"https://{ADLS_ACCOUNT_NAME}.dfs.core.windows.net",
           credential=CLIENT_SECRET
       )
       file_system_client = service_client.get_file_system_client(file_system=ADLS_CONTAINER_NAME)
       directory_client = file_system_client.get_directory_client(f"refresh-history/{datetime.now().strftime('%Y/%m/%d')}")
       directory_client.create_directory()
       file_client = directory_client.create_file(filename)
       file_client.upload_data(json.dumps(data), overwrite=True)
   
   # 主流程
   if __name__ == "__main__":
       token = get_pbi_token()
       datasets = get_datasets(token)
       
       for dataset in datasets:
           dataset_id = dataset["id"]
           dataset_name = dataset["name"]
           refresh_history = get_refresh_history(token, dataset_id)
           
           # 关联数据集信息到刷新历史
           for record in refresh_history:
               record["datasetId"] = dataset_id
               record["datasetName"] = dataset_name
               record["workspaceId"] = GROUP_ID
           
           # 上传到Data Lake
           upload_filename = f"{dataset_id}-refresh-history.json"
           upload_to_adls(refresh_history, upload_filename)
           print(f"已同步数据集 {dataset_name} 的刷新历史到Data Lake")
   

注意事项 #

• API请求限制：Power BI REST API有频次限制，批量调用时需控制速率，避免触发限流。
• 权限配置：确保服务主体同时拥有Power BI工作区的数据集读取权限，以及Data Lake的写入权限（如「Storage Blob Data Contributor」角色）。
• 数据格式：刷新历史返回的JSON包含刷新状态、起止时间、错误信息等字段，可根据需求清洗或转换。

内容的提问来源于stack exchange，提问作者bruce shavhani