如何在Databricks作业集群中通过笔记本安装Maven库

因为Maven库属于集群级依赖，不像Python库可以通过%pip在会话级安装，所以直接用常规方法无法在笔记本内完成安装。以下是几种无需修改ADF、仅通过笔记本即可实现的方案：

方案一：使用Databricks CLI在笔记本中执行安装命令 #

1. 先在笔记本中配置CLI环境（若集群未预配置）：

   import os
   # 建议通过机密范围获取令牌，避免硬编码
   os.environ['DATABRICKS_HOST'] = 'https://<你的工作区URL>'
   os.environ['DATABRICKS_TOKEN'] = dbutils.secrets.get("<机密范围名称>", "<令牌密钥名称>")
   

2. 获取当前作业集群的ID：

   from pyspark.sql import SparkSession
   spark = SparkSession.getActiveSession()
   cluster_id = spark.conf.get("spark.databricks.clusterUsageTags.clusterId")
   

3. 执行CLI命令添加Maven库：

   maven_coords = "com.example:your-library:1.0.0"  # 替换为你的Maven坐标
   os.system(f'databricks libraries install --cluster-id {cluster_id} --maven-coordinates {maven_coords}')
   

   注意：需确保作业集群实例已安装Databricks CLI，可提前在集群初始化脚本中完成CLI安装。

方案二：调用Databricks REST API安装Maven库 #

无需依赖CLI，直接通过API发送安装请求：

import requests
import json

# 配置参数
workspace_url = "https://<你的工作区URL>"
token = dbutils.secrets.get("<机密范围名称>", "<令牌密钥名称>")
cluster_id = spark.conf.get("spark.databricks.clusterUsageTags.clusterId")
maven_coords = "com.example:your-library:1.0.0"

# 构建请求体
payload = {
    "cluster_id": cluster_id,
    "libraries": [
        {
            "maven": {
                "coordinates": maven_coords,
                # 可选：若依赖不在默认仓库，添加自定义仓库地址
                "repo": "https://your-maven-repo.com"
            }
        }
    ]
}

# 发送安装请求
response = requests.post(
    f"{workspace_url}/api/2.0/libraries/install",
    headers={"Authorization": f"Bearer {token}"},
    json=payload
)

# 验证请求结果
if response.status_code == 200:
    print("Maven库安装请求已提交")
else:
    print(f"安装请求失败：{response.text}")

方案三：预定义集群策略自动安装Maven库 #

如果作业集群可指定策略，可提前配置策略实现自动安装：

1. 在Databricks UI中创建集群策略，添加spark.databricks.libraries.maven配置项，填入你的Maven坐标
2. 在ADF作业配置中指定使用该集群策略（仅需一次配置，无需修改流水线逻辑）
3. 作业集群启动时会自动加载策略中定义的Maven库，笔记本无需额外操作

注意事项 #

• 若作业集群是临时集群，每次启动都需要执行安装操作（方案一/二），建议将安装代码放在笔记本最开头
• 执行安装的身份（服务主体或用户）需拥有cluster-manager或admin权限，能对集群进行库安装操作
• Maven库提交安装后，需等待加载完成再执行依赖代码，可通过API查询安装状态：

  status_response = requests.get(
      f"{workspace_url}/api/2.0/libraries/cluster-status?cluster_id={cluster_id}",
      headers={"Authorization": f"Bearer {token}"}
  )
  print(json.dumps(status_response.json(), indent=2))
  

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