PySpark从临时视图传JSON调用Power BI API报BadRequest故障排查

作为Spark新手，我尝试调用Power BI Admin API的PostWorkspaceInfo端点，需要传入修改后的工作区列表作为请求体。已经实现了获取工作区GUID列表并分批处理的逻辑，但从临时视图WorkspaceBatches获取的BodyExpression字段内容传入requests.post的json参数时，返回BadRequest错误；但将该字段内容直接粘贴到json参数中，API调用却能正常执行。尝试过多种格式转换均无效，希望解决该问题。

具体现象 #

• 临时视图WorkspaceBatches的BodyExpression字段内容：

  {"workspaces": ["034c87da-bbbb-4df8-becc-555e5916d6fa","043654b6-7868-aaaa-bdad-b1cf9ad95f4c","073c7513-dddd-40c3-bb6e-48d0f0b00b42"]}
  

• 传入字段时返回的错误：

  {"error":{"code":"BadRequest","message":"Bad Request","details":[{"message":"Error converting value \"{\"workspaces\": [\"034c87da-bbbb-4df8-becc-555e5916d6fa\",\"043654b6-7868-aaaa-bdad-b1cf9ad95f4c\",\"073c7513-dddd-40c3-bb6e-48d0f0b00b42\"]}\" to type 'Microsoft.PowerBI.ServiceContracts.Api.RequiredWorkspaces'. Path '', line 1, position 3096.","target":"requiredWorkspaces"}]}}
  

• 直接写死JSON内容时API调用正常。

完整代码（缺失客户端ID和密钥） #

#Import necessary libraries
import msal
import requests
import json
import pandas as pd
from datetime import date, timedelta, time

#Set Client ID and Secret for Service Principal
client_id = ""
client_secret = ""
authority_url = "https://login.microsoftonline.com/northinsights.com"
scope = ["https://analysis.windows.net/powerbi/api/.default"]

activityDate = date.today()

#Use MSAL to grab token
app = msal.ConfidentialClientApplication(client_id, authority=authority_url, client_credential=client_secret)
result = app.acquire_token_for_client(scopes=scope)


#Set URL for changed workspaces
url = "https://api.powerbi.com/v1.0/myorg/admin/workspaces/modified"

#Get changed workspaces
if 'access_token' in result:
    access_token = result['access_token']
    header = {'Content-Type':'application/json', 'Authorization':f'Bearer {access_token}'}
    api_call = requests.get(url=url, headers=header)
    print(json.loads(api_call.text))

df1 = spark.createDataFrame(json.loads(api_call.text))
df1.createOrReplaceTempView("modifiedworkspaces")
display(df1)

%%sql
--Batch workspace ids into 100

create or replace temporary view WorkspaceBatches as
WITH WorkspaceRanked AS 
(
    SELECT ROW_NUMBER() OVER ( ORDER BY id) AS Rank
    , id
    FROM modifiedworkspaces
),
WorkspaceBatched AS
(
    SELECT ROW_NUMBER() OVER (PARTITION BY(Rank % 100) ORDER BY Rank ASC) AS Batch
    , id
    FROM WorkspaceRanked
)

SELECT replace(concat('{"workspaces": ["' , array_join(array_sort(collect_set(id)),","),'"}']),',','",""')  AS BodyExpression, Batch
--INTO WorkspaceBatches
FROM WorkspaceBatched
GROUP BY Batch;



SELECT * FROM WorkspaceBatches


#Try to post workspaces to get inventory
post_WorkspaceInfoUrl = "https://api.powerbi.com/v1.0/myorg/admin/workspaces/getInfo?lineage=True&datasourceDetails=True&getArtifactUsers=True&datasetSchema=True&datasetExpressions=True"
dfbatches = spark.sql("SELECT Batch, BodyExpression FROM WorkspaceBatches")
rows_looped = dfbatches.select("Batch", "BodyExpression").collect()
for rows in rows_looped:
    post_WorkspaceInfo = requests.post(url=post_WorkspaceInfoUrl, headers=header, json = rows.BodyExpression )
    print(post_WorkspaceInfo.text)

问题原因 #

从临时视图获取的BodyExpression是字符串类型的JSON，而requests.post的json参数需要传入Python字典对象。直接传入字符串时，requests会将其作为JSON字符串再次序列化，导致API收到的是嵌套的字符串（比如"{\"workspaces\": [...]}"），无法解析成要求的RequiredWorkspaces类型。

修复步骤 #

1. 解析字符串为字典：在传入json参数前，用json.loads()将BodyExpression的字符串内容转换成Python字典。
2. 优化SQL拼接逻辑（可选）：原SQL中的字符串拼接容易出现格式错误，建议改用Spark的内置JSON函数直接生成结构化JSON，而非手动拼接字符串，避免后续解析问题。

修改后的代码片段 #

修复POST请求部分 #

#Try to post workspaces to get inventory
post_WorkspaceInfoUrl = "https://api.powerbi.com/v1.0/myorg/admin/workspaces/getInfo?lineage=True&datasourceDetails=True&getArtifactUsers=True&datasetSchema=True&datasetExpressions=True"
dfbatches = spark.sql("SELECT Batch, BodyExpression FROM WorkspaceBatches")
rows_looped = dfbatches.select("Batch", "BodyExpression").collect()
for rows in rows_looped:
    # 将字符串JSON解析为字典
    payload = json.loads(rows.BodyExpression)
    post_WorkspaceInfo = requests.post(url=post_WorkspaceInfoUrl, headers=header, json=payload )
    print(post_WorkspaceInfo.text)

优化SQL拼接（可选推荐） #

将原SQL中的BodyExpression生成逻辑替换为使用to_json函数直接构造JSON结构，避免手动拼接出错：

create or replace temporary view WorkspaceBatches as
WITH WorkspaceRanked AS 
(
    SELECT ROW_NUMBER() OVER ( ORDER BY id) AS Rank
    , id
    FROM modifiedworkspaces
),
WorkspaceBatched AS
(
    SELECT ROW_NUMBER() OVER (PARTITION BY(Rank % 100) ORDER BY Rank ASC) AS Batch
    , id
    FROM WorkspaceRanked
)
SELECT to_json(named_struct('workspaces', array_sort(collect_set(id)))) AS BodyExpression, Batch
FROM WorkspaceBatched
GROUP BY Batch;

内容的提问来源于stack exchange，提问作者Ásgeir