概要

Azure Databricks に Azure Data Factory の Manged ID を Single user access として登録する方法を共有します。私が検証したところ、 GUI で設定することはできず、REST API で設定する必要があるようです。

Azure Data Factory のマネージドID を登録すると、Databricks 上では Service principals として管理されます。

クラスターの設定画面にて、Application ID をSingle user accessの欄には入力できなかったため、GUI では実施できないと判断しました。

本記事では、動作確認済み Python での設定例を共有します。

手順

1. 事前準備

import requests
import json

cluster_name = "adf's Cluster"  # クラスター名
single_user_name = ""  # シングルユーザーとして登録するサービスプリンシパルの ID

# トークンを取得
token = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()

# Databricks Workspace の URL をセット(想定通りに取得できない場合には手動でセット)
ws_url = "https://"
ws_url += spark.conf.get("spark.databricks.workspaceUrl")

2. クラスターを作成

# Authentication
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# API endpoint for cluster creation
endpoint = f"{ws_url}/api/2.0/clusters/create"

# Specify the configuration of the cluster to be created in JSON format
cluster_config = {
    "cluster_name": cluster_name,
    "spark_version": "12.2.x-scala2.12",
    "num_workers": "1",
    "node_type_id": "Standard_DS3_v2",
    "azure_attributes": {
        "first_on_demand": 1,
        "availability": "ON_DEMAND_AZURE",
        "spot_bid_max_price": -1
    },
    "autotermination_minutes": 30,
    "single_user_name": single_user_name,
    "data_security_mode": "LEGACY_SINGLE_USER_STANDARD",
    "runtime_engine": "STANDARD",
    "spark_conf": {
        "spark.databricks.delta.preview.enabled": "true",
    },
}

# Create a POST body for Cluster API by converting dictionary to JSON
body = json.dumps(cluster_config)

# Send a POST request to create the cluster
create_clusters_response = requests.post(endpoint, headers=headers, data=body)

# Print the output
print(create_clusters_response.json())

3. クラスターをピン止め

# Authentication
token = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# Specify the API endpoint for pinning the cluster
endpoint = f"{ws_url}/api/2.0/clusters/pin"

# Specify the configuration of the cluster to be pinned
pin_config = {
    "cluster_id": create_clusters_response.json()["cluster_id"],
}

# Create a POST body for Cluster API by converting dictionary to JSON
body = json.dumps(pin_config)

# Send a POST request to pin the cluster
pin_cluster_response = requests.post(endpoint, headers=headers, data=body)

# Print the output
print(pin_cluster_response.json())

4. クラスターの停止

# Authentication
token = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# Specify the API endpoint for pinning the cluster
endpoint = f"{ws_url}/api/2.0/clusters/delete"

# Specify the configuration of the cluster to be pinned
pin_config = {
    "cluster_id": create_clusters_response.json()["cluster_id"],
}

# Create a POST body for Cluster API by converting dictionary to JSON
body = json.dumps(pin_config)

# Send a POST request to pin the cluster
pin_cluster_response = requests.post(endpoint, headers=headers, data=body)

# Print the output
print(pin_cluster_response.json())

5. クラスターを確認

6. クラスターを削除

クラスター名の横にあるピン止めボタンを選択して、クラスターのピン止めを解除する。

Deleteを選択。Deleteが非アクティブの場合には、ピン止め設定されてないかを確認。