azure - 如何使用 Azure databricks API 创建 Python Wheel 类型的 Azure Databricks 作业

Question

我想使用 databricks API 在 Azure 中创建一个“pythonwheel”类型的 databricks 作业。我有一个 python 轮子，需要在这项工作中执行。

这个问题与我在 this stackoverflow link 的其他问题相关。 ，只是用于实现此目的的技术发生了变化。

遵循 Azure databricks API documentation我知道如何创建可以执行笔记本的数据 block 作业。但是，我需要的是类型“pythonwheel”的databricks工作。 我所有的代码都是在 python 轮子中实现的，它需要 24/7 运行。根据我从开发团队那里得到的要求，他们需要有一个“pythonwheel”类型的工作，而不是“notebook”类型。

如您所见databricks documentation已经展示了如何从 databricks 工作区创建 pythonwheel 类型的作业。然而，我需要在 DevOps 管道中自动化此过程，这就是为什么我想通过对 databricks API 进行 API 调用来实现这一点。下面是我为创建 databricks 作业而实现的代码。此代码使用笔记本来执行代码。正如我提到的，我需要运行一个“pythonwheel”作业，正如它所解释的那样here 。您可以在下面看到工作区中的此类作业:

我当前的代码如下:我的目标是将其更改为运行Python轮而不是笔记本:

import requests
import os


# both 2.0 and 2.1 API can create job.
dbrks_create_job_url = "https://"+os.environ['DBRKS_INSTANCE']+".azuredatabricks.net/api/2.1/jobs/create"

DBRKS_REQ_HEADERS = {
    'Authorization': 'Bearer ' + os.environ['DBRKS_BEARER_TOKEN'],
    'X-Databricks-Azure-Workspace-Resource-Id': '/subscriptions/'+ os.environ['DBRKS_SUBSCRIPTION_ID'] +'/resourceGroups/'+ os.environ['DBRKS_RESOURCE_GROUP'] +'/providers/Microsoft.Databricks/workspaces/' + os.environ['DBRKS_WORKSPACE_NAME'],
    'X-Databricks-Azure-SP-Management-Token': os.environ['DBRKS_MANAGEMENT_TOKEN']}

CLUSTER_ID = "\"" + os.environ["DBRKS_CLUSTER_ID"] + "\""
NOTEBOOK_LOCATION = "\"" + os.environ["NOTEBOOK_LOCATION"] + "test-notebook" + "\""
print("Notebook path is {}".format(NOTEBOOK_LOCATION))
print(CLUSTER_ID)

body_json = """
    {
    "name": "A sample job to trigger from DevOps",
    "tasks": [
        {
        "task_key": "ExecuteNotebook",
        "description": "Execute uploaded notebook including tests",
        "depends_on": [],
        "existing_cluster_id": """ + CLUSTER_ID + """,
        "notebook_task": {
          "notebook_path": """ + NOTEBOOK_LOCATION + """,
          "base_parameters": {}
        },
        "timeout_seconds": 300,
        "max_retries": 1,
        "min_retry_interval_millis": 5000,
        "retry_on_timeout": false
      }
],
    "email_notifications": {},
    "name": "Run_Unit_Tests",
    "max_concurrent_runs": 1}
"""

print("Request body in json format:")
print(body_json)

response = requests.post(dbrks_create_job_url, headers=DBRKS_REQ_HEADERS, data=body_json) 

if response.status_code == 200:
    print("Job created successfully!")
    print(response.status_code)
    print(response.content)
    print("Job Id = {}".format(response.json()['job_id']))
    print("##vso[task.setvariable variable=DBRKS_JOB_ID;isOutput=true;]{b}".format(b=response.json()['job_id'])) 
else:
    print("job failed!")
    raise Exception(response.content)

Answer 1

正如 @Alex Ott 已经提到的，您需要使用 pyhton_wheel_task，而不是使用 notebook_task。

基于Job API 2.1 docs .