Python용 Databricks SDK

이 문서에서는 Azure Databricks 작업을 자동화하고 Python용 Databricks SDK를 사용하여 개발을 가속화하는 방법을 알아봅니다. 이 문서에서는 Read The Docs에 대한 Python용 Databricks SDK 설명서 와 GitHub의 Python용 Databricks SDK 리포지토리의 코드 예제를 보완합니다.

시작하기 전에

Azure Databricks Notebook 내에서 또는 로컬 개발 컴퓨터에서 Python용 Databricks SDK를 사용할 수 있습니다.

• Azure Databricks Notebook 내에서 Python용 Databricks SDK를 사용하려면 Azure Databricks Notebook에서 Python용 Databricks SDK 사용으로 건너뜁니다.
• 로컬 개발 컴퓨터에서 Python용 Databricks SDK를 사용하려면 이 섹션의 단계를 완료합니다.

Python용 Databricks SDK를 사용하기 전에 개발 머신은 다음 요건을 충족해야 합니다.

• Azure Databricks 인증 구성
• Python 3.8 이상이 설치되어 있습니다. Azure Databricks 컴퓨팅 리소스를 자동화하기 위해 Databricks는 대상 Azure Databricks 컴퓨팅 리소스에 설치된 버전과 일치하는 Python의 주 버전과 부 버전을 설치하는 것이 좋습니다. 이 문서의 예제에서는 Python 3.10이 설치된 Databricks Runtime 13.3 LTS를 사용하여 클러스터를 자동화합니다. 올바른 버전은 클러스터의 Databricks Runtime 버전에 대한 Databricks Runtime 릴리스 정보 버전 및 호환성을 참조하세요.
• Databricks에서는 Python용 Databricks SDK와 함께 사용하는 각 Python 프로젝트에 대해 Python 가상 환경을 만들고 활성화하는 것이 좋습니다. Python 가상 환경은 코드 프로젝트에서 호환되는 버전의 Python 및 Python 패키지(이 경우 Python 패키지용 Databricks SDK)를 사용하는지 확인하는 데 도움이 됩니다. Python 가상 환경에 대한 자세한 내용은 venv 또는 Poetry를 참조하세요.

Python용 Databricks SDK 시작

이 섹션에서는 로컬 개발 컴퓨터에서 Python용 Databricks SDK를 시작하는 방법을 설명합니다. Azure Databricks Notebook 내에서 Python용 Databricks SDK를 사용하려면 Azure Databricks Notebook에서 Python용 Databricks SDK 사용으로 건너뜁니다.

1. Azure Databricks 인증이 구성된 개발 머신에서 Python이 이미 설치되어 있고 Python 가상 환경이 이미 활성화되어 있는 경우 다음과 같이 Python 패키지 인덱스(PyPI)에서 databricks-sdk 패키지(및 해당 종속성)를 설치합니다.

   Venv

   pip를 사용하여 databricks-sdk 패키지를 설치합니다. (일부 시스템에서는 여기와 전체에서 pip3을 pip로 교체해야 할 수도 있습니다.)

   pip3 install databricks-sdk
   

   Poetry

   poetry add databricks-sdk
   

   Python용 Databricks SDK가 베타에 있는 동안 특정 버전의 databricks-sdk 패키지를 설치하려면 패키지의 릴리스 기록을 참조하세요. 예를 들어 버전 0.1.6을 설치하려면 다음을 수행합니다.

   Venv

   pip3 install databricks-sdk==0.1.6
   

   Poetry

   poetry add databricks-sdk==0.1.6
   

   Python 패키지용 Databricks SDK의 기존 설치를 최신 버전으로 업데이트하려면 다음 명령을 사용합니다.

   Venv

   pip3 install --upgrade databricks-sdk
   

   Poetry

   poetry add databricks-sdk@latest
   

   Python 패키지의 현재 Version 및 기타 세부 정보에 대한 Databricks SDK를 표시하려면 다음 명령을 실행합니다.

   Venv

   pip3 show databricks-sdk
   

   Poetry

   poetry show databricks-sdk
   

2. Python 가상 환경에서 Python용 Databricks SDK를 가져오는 Python 코드 파일을 만듭니다. 다음 예제에서는 다음 내용이 포함된 main.py라는 파일에서 Azure Databricks 작업 영역의 모든 클러스터를 나열합니다.

   from databricks.sdk import WorkspaceClient
   
   w = WorkspaceClient()
   
   for c in w.clusters.list():
     print(c.cluster_name)
   

3. python 명령을 실행하여 main.py라는 파일을 가정한 Python 코드 파일을 실행합니다.

   Venv

   python3.10 main.py
   

   Poetry

   가상 환경의 셸에 있는 경우.

   python3.10 main.py
   

   가상 환경의 셸에 없는 경우.

   poetry run python3.10 main.py
   

   참고 항목

   이전 호출에서 어떤 인수도 w = WorkspaceClient()로 설정하지 않음으로써 Python용 Databricks SDK는 Azure Databricks 인증을 수행하기 위해 기본 프로세스를 사용합니다. 이 기본 동작을 재정의하려면 다음 인증 섹션을 참조하세요.

Azure Databricks 계정 또는 작업 영역을 사용하여 Python용 Databricks SDK 인증

이 섹션에서는 로컬 개발 컴퓨터에서 Azure Databricks 계정 또는 작업 영역으로 Python용 Databricks SDK를 인증하는 방법을 설명합니다. Azure Databricks Notebook 내에서 Python용 Databricks SDK를 인증하려면 Azure Databricks Notebook에서 Python용 Databricks SDK 사용으로 건너뜁니다.

Python용 Databricks SDK는 통합되고 일관된 아키텍처 및 프로그래밍 방식의 인증 방식인 Databricks 클라이언트 통합 인증 표준을 구현합니다. 이 방법을 사용하면 Azure Databricks를 사용하여 보다 중앙 집중화하고 예측 가능하게 인증을 설정하고 자동화할 수 있습니다. 이를 통해 Databricks 인증을 한 번 구성한 다음, 추가 인증 구성 변경 없이 여러 Databricks 도구 및 SDK에서 해당 구성을 사용할 수 있습니다. Python의 전체 코드 예제를 비롯한 자세한 내용은 Databricks 클라이언트 통합 인증을 참조하세요.

Python용 Databricks SDK를 사용하여 Databricks 인증을 초기화하는 데 사용할 수 있는 코딩 패턴 중 일부는 다음과 같습니다.

• 다음 중 하나를 수행하여 Databricks 기본 인증을 사용합니다.

  • 대상 Databricks 인증 유형에 필요한 필드를 사용하여 사용자 지정 Databricks 구성 프로필을 만들거나 식별합니다. 그런 다음 DATABRICKS_CONFIG_PROFILE 환경 변수를 사용자 지정 구성 프로필의 이름으로 설정합니다.
  • 대상 Databricks 인증 유형에 필요한 환경 변수를 설정합니다.

  그런 다음, 다음과 같이 Databricks 기본 인증을 사용하여 WorkspaceClient 개체를 인스턴스화합니다.

  from databricks.sdk import WorkspaceClient
  
  w = WorkspaceClient()
  # ...
  

• Azure Databricks 개인용 액세스 토큰과 같은 중요한 정보를 코드에 노출할 위험이 있으므로 필수 필드를 하드 코딩하는 것은 지원되지만 권장되지는 않습니다. 다음 예제에서는 Databricks 토큰 인증에 대한 Azure Databricks 호스트 및 액세스 토큰 값을 하드 코딩합니다.

  from databricks.sdk import WorkspaceClient
  
  w = WorkspaceClient(
    host  = 'https://...',
    token = '...'
  )
  # ...
  

Python용 Databricks SDK 설명서의 인증도 참조하세요.

Azure Databricks Notebook에서 Python용 Databricks SDK 사용

Python용 Databricks SDK가 설치된 연결된 Azure Databricks 클러스터가 있는 Azure Databricks Notebook에서 Python용 Databricks SDK 기능을 호출할 수 있습니다. Databricks Runtime 13.3 LTS 이상을 사용하는 모든 Azure Databricks 클러스터에 기본적으로 설치되어 있습니다. Databricks Runtime 12.2 LTS 이하를 사용하는 Azure Databricks 클러스터의 경우 먼저 Python용 Databricks SDK를 설치해야 합니다. 1단계: Python용 Databricks SDK 설치 또는 업그레이드를 참조하세요.

특정 Databricks Runtime 버전에 설치된 Python용 Databricks SDK 버전을 확인하려면 해당 버전에 대한 Databricks Runtime 릴리스 정보의 설치된 Python 라이브러리 섹션을 참조하세요.

Databricks는 최신 SDK 버전을 PiPy에서 설치할 것을 권장합니다. 최소한 Python용 Databricks SDK 0.6.0 이상을 설치 또는 업그레이드해야 합니다. 기본 Azure Databricks Notebook 인증은 Databricks Runtime의 모든 버전에서 0.6.0 이상부터 지원됩니다.

다음 표는 Python 및 Databricks Runtime 버전용 Databricks SDK에 대한 Notebook 인증 지원을 설명합니다.

기본 Azure Databricks Notebook 인증은 Azure Databricks가 자체 사용을 위해 백그라운드에서 자동으로 생성하는 임시 Azure Databricks 개인용 액세스 토큰을 사용합니다. Azure Databricks는 Notebook 실행이 중지된 후 이 임시 토큰을 삭제합니다.

Azure Databricks 계정 수준 API를 호출하거나 기본 Databricks Notebook 인증 이외의 Databricks 인증 유형을 사용하려는 경우 다음 인증 유형도 지원됩니다.

Azure 관리 ID 인증이 지원되지 않습니다.

1단계: Python용 Databricks SDK 설치 또는 업그레이드

1. Azure Databricks Python Notebook은 다른 Python 라이브러리와 마찬가지로 Python용 Databricks SDK를 사용할 수 있습니다. 연결된 Azure Databricks 클러스터에 Python용 Databricks SDK 라이브러리를 설치하거나 업그레이드하려면 다음과 같이 Notebook 셀에서 %pip 매직 명령을 실행합니다.

   %pip install databricks-sdk --upgrade
   

2. %pip 매직 명령을 실행한 후에는 Python을 다시 시작하여 설치된 라이브러리 또는 업그레이드된 라이브러리를 Notebook에서 사용할 수 있도록 해야 합니다. 이렇게 하려면 %pip 매직 명령을 사용하여 셀 바로 뒤의 Notebook 셀에서 다음 명령을 실행합니다.

   dbutils.library.restartPython()
   

3. Python용 Databricks SDK의 설치된 버전을 표시하려면 Notebook 셀에서 다음 명령을 실행합니다.

   %pip show databricks-sdk | grep -oP '(?<=Version: )\S+'
   

2단계: 코드 실행

Notebook 셀에서 Python용 Databricks SDK를 가져온 다음 호출하는 Python 코드를 만듭니다. 다음 예제에서는 기본 Azure Databricks Notebook 인증을 사용하여 Azure Databricks 작업 영역의 모든 클러스터를 나열합니다.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

for c in w.clusters.list():
  print(c.cluster_name)

이 셀을 실행하면 Azure Databricks 작업 영역에서 사용 가능한 모든 클러스터의 이름 목록이 나타납니다.

다른 Azure Databricks 인증 유형을 사용하려면 Azure Databricks 인증 방법을 참조하고 해당 링크를 클릭하여 추가 기술 세부 정보를 확인하세요.

Databricks 유틸리티 사용

로컬 개발 컴퓨터 또는 Azure Databricks Notebook 내에서 실행되는 Python 코드용 Databricks SDK에서 Databricks 유틸리티(dbutils) 참조를 호출할 수 있습니다.

• 로컬 개발 컴퓨터에서 Databricks 유틸리티는 dbutils.fs, dbutils.secrets, dbutils.widgets 및 dbutils.jobs 명령 그룹에만 액세스할 수 있습니다.
• Azure Databricks 클러스터에 연결된 Azure Databricks Notebook에서 Databricks Utilities는 dbutils.fs, dbutils.secrets 및 dbutils.widgets뿐만 아니라 사용 가능한 모든 Databricks Utilities 명령 그룹에 액세스할 수 있습니다. 또한 dbutils.notebook 명령 그룹은 두 가지 수준의 명령으로만 제한됩니다(예: dbutils.notebook.run 또는 dbutils.notebook.exit).

로컬 개발 컴퓨터 또는 Azure Databricks Notebook에서 Databricks 유틸리티를 호출하려면 WorkspaceClient 내의 dbutils를 사용합니다. 이 코드 예제에서는 기본 Azure Databricks Notebook 인증을 사용하여 작업 영역의 DBFS 루트에 있는 모든 개체의 경로를 나열하기 위해 WorkspaceClient 내의 dbutils를 호출합니다.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
d = w.dbutils.fs.ls('/')

for f in d:
  print(f.path)

또는 직접 dbutils를 호출할 수 있습니다. 그러나 기본 Azure Databricks Notebook 인증만 사용하도록 제한됩니다. 이 코드 예제에서는 작업 영역의 DBFS 루트에 있는 모든 개체를 나열하기 위해 직접 dbutils를 호출합니다.

from databricks.sdk.runtime import *

d = dbutils.fs.ls('/')

for f in d:
  print(f.path)

Unity 카탈로그 볼륨에 액세스하려면 WorkspaceClient 내의 files을 사용합니다. Unity 카탈로그 볼륨의 파일 관리를 참조하세요. 볼륨에 액세스하기 위해 dbutils 단독으로나 WorkspaceClient 내에서 사용할 수 없습니다.

dbutils와의 상호 작용도 참조하세요.

코드 예제

다음 코드 예제에서는 Python용 Databricks SDK를 사용하여 클러스터를 만들고 삭제하며, 작업을 실행하고, 계정 수준 그룹을 나열하는 방법을 보여 줍니다. 이러한 코드 예제는 기본 Azure Databricks Notebook 인증을 사용합니다. 기본 Azure Databricks Notebook 인증에 대한 세부 정보는 Azure Databricks Notebook에서 Python용 Databricks SDK 사용을 참조하세요. Notebook 외부의 기본 인증에 대한 세부 정보는 Azure Databricks 계정 또는 작업 영역으로 Python용 Databricks SDK 인증을 참조하세요.

추가 코드 예제는 GitHub의 Python용 Databricks SDK 리포지토리의 예제 폴더를 참조하세요. 다음을 참조하세요.

• Databricks 작업 영역 API 참조

• Databricks 계정 API 참조

• 클러스터 만들기

• 클러스터 영구 삭제

• 작업 만들기

• 서버리스 컴퓨팅을 사용하는 작업 만들기

• Unity 카탈로그 볼륨의 파일 관리

• 계정 수준 그룹 나열

클러스터 생성

이 코드 예제에서는 지정된 Databricks Runtime 버전 및 클러스터 노드 형식을 사용하여 클러스터를 만듭니다. 이 클러스터에는 하나의 작업자가 있으며 클러스터는 15분의 유휴 시간 후에 자동으로 종료됩니다.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

print("Attempting to create cluster. Please wait...")

c = w.clusters.create_and_wait(
  cluster_name             = 'my-cluster',
  spark_version            = '12.2.x-scala2.12',
  node_type_id             = 'Standard_DS3_v2',
  autotermination_minutes  = 15,
  num_workers              = 1
)

print(f"The cluster is now ready at " \
      f"{w.config.host}#setting/clusters/{c.cluster_id}/configuration\n")

클러스터 영구 삭제

이 코드 예제에서는 작업 영역에서 지정된 클러스터 ID를 사용하여 클러스터를 영구적으로 삭제합니다.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

c_id = input('ID of cluster to delete (for example, 1234-567890-ab123cd4): ')

w.clusters.permanent_delete(cluster_id = c_id)

작업 만들기

이 코드 예제에서는 지정된 클러스터에서 지정된 Notebook을 실행하는 Azure Databricks 작업을 만듭니다. 코드가 실행되면 터미널의 사용자로부터 기존 Notebook의 경로, 기존 클러스터 ID 및 관련 작업 설정을 가져옵니다.

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import Task, NotebookTask, Source

w = WorkspaceClient()

job_name            = input("Some short name for the job (for example, my-job): ")
description         = input("Some short description for the job (for example, My job): ")
existing_cluster_id = input("ID of the existing cluster in the workspace to run the job on (for example, 1234-567890-ab123cd4): ")
notebook_path       = input("Workspace path of the notebook to run (for example, /Users/someone@example.com/my-notebook): ")
task_key            = input("Some key to apply to the job's tasks (for example, my-key): ")

print("Attempting to create the job. Please wait...\n")

j = w.jobs.create(
  name = job_name,
  tasks = [
    Task(
      description = description,
      existing_cluster_id = existing_cluster_id,
      notebook_task = NotebookTask(
        base_parameters = dict(""),
        notebook_path = notebook_path,
        source = Source("WORKSPACE")
      ),
      task_key = task_key
    )
  ]
)

print(f"View the job at {w.config.host}/#job/{j.job_id}\n")

서버리스 컴퓨팅을 사용하는 작업 만들기

다음 예제에서는 작업용 서버리스 컴퓨팅을 사용하는 작업을 만듭니다.

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import NotebookTask, Source, Task

w = WorkspaceClient()

j = w.jobs.create(
  name = "My Serverless Job",
  tasks = [
    Task(
      notebook_task = NotebookTask(
      notebook_path = "/Users/user@databricks.com/MyNotebook",
      source = Source("WORKSPACE")
      ),
      task_key = "MyTask",
   )
  ]
)

Unity 카탈로그 볼륨의 파일 관리

이 코드 예제에서는 Unity 카탈로그 볼륨에 액세스하기 위해 WorkspaceClient 내 files 기능에 대한 다양한 호출을 보여 줍니다.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Define volume, folder, and file details.
catalog            = 'main'
schema             = 'default'
volume             = 'my-volume'
volume_path        = f"/Volumes/{catalog}/{schema}/{volume}" # /Volumes/main/default/my-volume
volume_folder      = 'my-folder'
volume_folder_path = f"{volume_path}/{volume_folder}" # /Volumes/main/default/my-volume/my-folder
volume_file        = 'data.csv'
volume_file_path   = f"{volume_folder_path}/{volume_file}" # /Volumes/main/default/my-volume/my-folder/data.csv
upload_file_path   = './data.csv'

# Create an empty folder in a volume.
w.files.create_directory(volume_folder_path)

# Upload a file to a volume.
with open(upload_file_path, 'rb') as file:
  file_bytes = file.read()
  binary_data = io.BytesIO(file_bytes)
  w.files.upload(volume_file_path, binary_data, overwrite = True)

# List the contents of a volume.
for item in w.files.list_directory_contents(volume_path):
  print(item.path)

# List the contents of a folder in a volume.
for item in w.files.list_directory_contents(volume_folder_path):
  print(item.path)

# Print the contents of a file in a volume.
resp = w.files.download(volume_file_path)
print(str(resp.contents.read(), encoding='utf-8'))

# Delete a file from a volume.
w.files.delete(volume_file_path)

# Delete a folder from a volume.
w.files.delete_directory(volume_folder_path)

계정 수준 그룹 나열

이 코드 예제에서는 Azure Databricks 계정 내에서 사용 가능한 모든 그룹의 표시 이름을 나열합니다.

from databricks.sdk import AccountClient

a = AccountClient()

for g in a.groups.list():
  print(g.display_name)

테스트

코드를 테스트하려면 pytest와 같은 Python 테스트 프레임워크를 사용합니다. Azure Databricks REST API 엔드포인트를 호출하지 않거나 Azure Databricks 계정 또는 작업 영역의 상태를 변경하지 않고 시뮬레이션된 조건에서 코드를 테스트하려면 unittest.mock와 같은 Python 모의 라이브러리를 사용합니다.

다음 helpers.py 예제 파일에는 새 클러스터에 create_cluster 대한 정보를 반환하는 함수가 포함되어 있습니다.

# helpers.py

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.compute import ClusterDetails

def create_cluster(
  w: WorkspaceClient,
  cluster_name:            str,
  spark_version:           str,
  node_type_id:            str,
  autotermination_minutes: int,
  num_workers:             int
) -> ClusterDetails:
  response = w.clusters.create(
    cluster_name            = cluster_name,
    spark_version           = spark_version,
    node_type_id            = node_type_id,
    autotermination_minutes = autotermination_minutes,
    num_workers             = num_workers
  )
  return response

함수를 호출하는 명명된 main.py 다음 파일을 지정합니다 create_cluster .

# main.py

from databricks.sdk import WorkspaceClient
from helpers import *

w = WorkspaceClient()

# Replace <spark-version> with the target Spark version string.
# Replace <node-type-id> with the target node type string.
response = create_cluster(
  w = w,
  cluster_name            = 'Test Cluster',
  spark_version           = '<spark-version>',
  node_type_id            = '<node-type-id>',
  autotermination_minutes = 15,
  num_workers             = 1
)

print(response.cluster_id)

test_helpers.py(으)로 명명된 다음 파일은 create_cluster 함수가 예상 응답을 반환하는지 여부를 테스트합니다. 이 테스트는 대상 작업 영역에 클러스터를 만드는 대신 WorkspaceClient 개체를 모의하고, 모의 개체의 설정을 정의한 다음 모의 개체를 create_cluster 함수에 전달합니다. 그런 다음 테스트를 통해 함수가 새 모의 클러스터의 예상 ID를 반환하는지 여부를 검사합니다.

# test_helpers.py

from databricks.sdk import WorkspaceClient
from helpers import *
from unittest.mock import create_autospec # Included with the Python standard library.

def test_create_cluster():
  # Create a mock WorkspaceClient.
  mock_workspace_client = create_autospec(WorkspaceClient)

  # Set the mock WorkspaceClient's clusters.create().cluster_id value.
  mock_workspace_client.clusters.create.return_value.cluster_id = '123abc'

  # Call the actual function but with the mock WorkspaceClient.
  # Replace <spark-version> with the target Spark version string.
  # Replace <node-type-id> with the target node type string.
  response = create_cluster(
    w = mock_workspace_client,
    cluster_name            = 'Test Cluster',
    spark_version           = '<spark-version>',
    node_type_id            = '<node-type-id>',
    autotermination_minutes = 15,
    num_workers             = 1
  )

  # Assert that the function returned the mocked cluster ID.
  assert response.cluster_id == '123abc'

이 테스트를 실행하려면 코드 프로젝트의 루트에서 pytest 명령을 실행하며, 이 명령은 다음과 유사한 테스트 결과를 생성해야 합니다.

$ pytest
=================== test session starts ====================
platform darwin -- Python 3.12.2, pytest-8.1.1, pluggy-1.4.0
rootdir: <project-rootdir>
collected 1 item

test_helpers.py . [100%]
======================== 1 passed ==========================

추가 리소스

자세한 내용은 다음을 참조하세요.

• Python용 Databricks SDK 설명서

• 추가 코드 예제

• Databricks 작업 영역 API 참조

• Databricks 계정 API 참조

• 로깅

• 장기 실행 작업

• 페이지 매김 응답

• OAuth를 사용하여 Single Sign-On