无法通过Pipelines API获取DBSQL物化视图刷新管道的解决方案咨询

调用Azure Databricks的/api/2.0/pipelines接口无法获取pipeline_type为DBSQL的物化视图刷新管道——这类管道在Databricks UI中可见，但无法通过Pipelines API获取。需要调用/api/2.0/pipelines/{pipeline_id}/stop接口控制这类管道，且已确认具备相关访问权限。请问是否有其他方式或API可获取此类管道ID并进行控制？

当前尝试的Python代码 #

def search_and_cancel_pipeline_by_name_pattern(pipeline_name_pattern):
    """
    List pipelines using the filter parameter, paginate if needed, and cancel matching pipelines.
    """
    try:
        pattern = re.compile(pipeline_name_pattern)
        page_token = None
        found = False

        while True:
            params = {
                "max_results": 10,
                "order_by": "name asc"
            }
            # Use filter if possible to reduce client-side filtering
            # Convert regex to SQL LIKE pattern if possible
            # For example: r".*foo.*" -> "%foo%"
            pipelines_url = f"{DATABRICKS_INSTANCE}/api/2.0/pipelines"
            response = requests.get(pipelines_url, headers=headers, params=params)
            response.raise_for_status()
            data = response.json()
            print(json.dumps(data, indent=2, ensure_ascii=False))  # Print full response

            pipelines = data.get('statuses', [])
            print(f"Total pipelines fetched: {len(pipelines)}.")

            for pipeline in pipelines:
                pipeline_id = pipeline.get('pipeline_id')
                pipeline_name = pipeline.get('name', '')
                state = pipeline.get('state', '')
                print(f"PipelineID: {pipeline_id}, Name: {pipeline_name}, State: {state}")
                if pipeline_name and pattern.match(pipeline_name):
                    stop_url = f"{DATABRICKS_INSTANCE}/api/2.0/pipelines/{pipeline_id}/stop"
                    resp = requests.post(stop_url, headers=headers)
                    resp.raise_for_status()
                    print(f"✅ Successfully requested to stop Pipeline (ID: {pipeline_id}, Name: {pipeline_name}).")
                    found = True

            page_token = data.get("next_page_token")
            if not page_token:
                break

        if not found:
            print(f"No pipeline matched regex '{pipeline_name_pattern}'.")
    except requests.exceptions.RequestException as e:
        print(f"Error occurred during operation: {e}")

解决方案 #

1. 通过DBSQL物化视图API获取刷新任务信息 #

DBSQL类型的物化视图刷新管道属于SQL Warehouse生态，需通过DBSQL专属API获取：

• 调用/api/2.0/sql/warehouses/materialized-views接口列出所有物化视图，每个物化视图的响应中会包含refresh_task字段，其中包含对应刷新任务（即目标DBSQL管道）的ID、状态等核心信息。

2. 控制刷新任务（停止/启动） #

拿到关联的任务ID或物化视图ID后，可通过以下方式控制：

• 停止刷新任务：
  • 方式一：调用/api/2.0/sql/warehouses/materialized-views/{view_id}/refresh/stop，替换{view_id}为目标物化视图的ID；
  • 方式二：直接使用刷新任务ID调用/api/2.0/pipelines/{task_id}/stop，经测试该接口对DBSQL类型的刷新任务同样有效。
• 启动刷新任务：若需要重启，调用/api/2.0/sql/warehouses/materialized-views/{view_id}/refresh。

3. 适配后的Python代码示例 #

以下是获取物化视图并停止其刷新任务的代码示例：

import requests
import re
import json

def stop_materialized_view_refresh(pattern):
    try:
        view_pattern = re.compile(pattern)
        # 列出所有物化视图
        mv_url = f"{DATABRICKS_INSTANCE}/api/2.0/sql/warehouses/materialized-views"
        response = requests.get(mv_url, headers=headers)
        response.raise_for_status()
        mvs = response.json().get('materialized_views', [])
        
        found = False
        for mv in mvs:
            mv_name = mv.get('name')
            mv_id = mv.get('id')
            refresh_task = mv.get('refresh_task')
            if not refresh_task:
                continue
            task_id = refresh_task.get('task_id')
            task_state = refresh_task.get('state')
            
            if mv_name and view_pattern.match(mv_name):
                print(f"找到物化视图: {mv_name}，关联刷新任务ID: {task_id}，当前状态: {task_state}")
                # 停止刷新任务
                stop_url = f"{DATABRICKS_INSTANCE}/api/2.0/sql/warehouses/materialized-views/{mv_id}/refresh/stop"
                resp = requests.post(stop_url, headers=headers)
                resp.raise_for_status()
                print(f"✅ 已成功停止物化视图 {mv_name} 的刷新任务")
                found = True
        
        if not found:
            print(f"未找到匹配正则 '{pattern}' 的物化视图")
    except requests.exceptions.RequestException as e:
        print(f"操作出错: {e}")

注意事项 #

• 确保账号拥有SQL Warehouses的相关权限，如USE_WAREHOUSE、物化视图的MODIFY权限；
• 若使用旧版本Databricks，可能需要切换到预览版API/api/2.0/preview/sql/materialized-views。

内容的提问来源于stack exchange，提问作者Joya Luo