Python용 Databricks SDK

이 문서에서는 Python용 Databricks SDK를 사용하여 Azure Databricks 계정, 작업 영역 및 관련 리소스에서 작업을 자동화하는 방법을 알아봅니다. 이 문서에서는 Read The Docs에 대한 Python용 Databricks SDK 설명서 와 GitHub의 Python용 Databricks SDK 리포지토리의 코드 예제를 보완합니다.

참고 항목

이 기능은 베타 버전이며 프로덕션 환경에서도 사용할 수 있습니다.

베타 기간 동안 Databricks에서는 코드가 의존하는 Python용 Databricks SDK의 특정 부 버전에 대한 종속성을 고정하는 것이 좋습니다. 예를 들어 venv에 대해 requirements.txt, 또는 시에 대해 pyproject.tomlpoetry.lock 같은 파일에 종속성을 고정할 수 있습니다. 종속성을 고정하는 방법에 대한 자세한 내용은 venv를 위한 가상 환경 및 패키지 또는 시를 위한 종속성 설치를 참조하세요.

시작하기 전에

Azure Databricks Notebook 내에서 또는 로컬 개발 컴퓨터에서 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 또는 를 사용하는 방법을 설명합니다.

venv로 Python 가상 환경 만들기

  1. 터미널에서 Python 코드 프로젝트의 루트 디렉터리로 설정하여 다음 명령을 실행합니다. 이 명령은 venv에 가상 환경에 Python 3.10을 사용하도록 지시한 다음 Python 코드 프로젝트의 루트 디렉터리 내에 .venv라는 이름의 숨겨진 디렉터리에 가상 환경의 지원 파일을 만듭니다.

    # Linux and macOS
    python3.10 -m venv ./.venv
    
    # Windows
    python3.10 -m venv .\.venv
    
  2. venv을 사용하여 가상 환경을 활성화합니다. 운영 체제 및 터미널 유형에 따라 사용할 올바른 명령에 대한 venv 설명서를 참조하세요. 예를 들어 macOS에서 zsh를 실행합니다.

    source ./.venv/bin/activate
    

    가상 환경의 이름(예: .venv)이 터미널 프롬프트 바로 앞에 괄호로 표시되면 가상 환경이 활성화된다는 것을 알 수 있습니다.

    Python 가상 환경을 비활성화하려면 deactivate 명령을 실행합니다.

    가상 환경의 이름이 터미널 프롬프트 바로 앞에 괄호로 더 이상 표시되지 않으면 가상 환경이 비활성화된다는 것을 알 수 있습니다.

Python용 Databricks SDK를 시작으로 건너뜁니다.

시를 사용하여 가상 환경 만들기

  1. 아직 설치하지 않은 경우 시를 설치합니다.

  2. 터미널에서 Python 코드 프로젝트의 루트 디렉터리로 설정하고 다음 명령을 실행하여 시용 Python 코드 프로젝트를 초기화하도록 poetry에게 지시합니다.

    poetry init
    
  3. 시는 완료할 수 있는 몇 가지 프롬프트를 표시합니다. 이러한 프롬프트는 Python용 Databricks SDK에만 해당되는 것은 아닙니다. 이러한 프롬프트에 대한 자세한 내용은 init를 참조하세요.

  4. 프롬프트를 완료하면 시가 Python 프로젝트에 pyproject.toml 파일을 추가합니다. pyproject.toml 파일에 대한 자세한 내용은 pyproject.toml 파일을 참조하세요.

  5. 터미널을 계속 Python 코드 프로젝트의 루트 디렉터리로 설정하여 다음 명령을 실행합니다. 이 명령은 poetry에게 pyproject.toml 파일을 읽고, 종속성을 설치 및 해결하며, 종속성을 잠그는 poetry.lock 파일을 만들고, 마지막으로 가상 환경을 만들도록 지시합니다.

    poetry install
    
  6. 터미널에서 Python 코드 프로젝트의 루트 디렉터리로 설정하고 다음 명령을 실행하여 poetry에게 가상 환경을 활성화하고 셸을 입력하도록 지시합니다.

    poetry shell
    

    가상 환경의 이름이 터미널 프롬프트 바로 앞에 괄호로 표시되면 가상 환경이 활성화되고 셸이 입력된다는 것을 알 수 있습니다.

    가상 환경을 비활성화하고 언제라도 셸을 끝내려면 exit 명령을 실행합니다.

    가상 환경의 이름이 터미널 프롬프트 바로 앞에 괄호로 더 이상 표시되지 않으면 셸이 종료되었다는 것을 알 수 있습니다.

    시 가상 환경을 만들고 관리하는 방법에 대한 자세한 내용은 환경 관리를 참조하세요.

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 패키지를 설치합니다. (일부 시스템에서는 여기와 전체에서 pip3pip로 교체해야 할 수도 있습니다.)

    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는 아직 Azure 관리 ID 인증을 구현하지 않았습니다.

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 기능을 호출할 수 있습니다. 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 버전에 대한 Databricks Runtime 릴리스 정보에서 "설치된 Python 라이브러리" 섹션을 참조하세요.

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

Important

기본 Azure Databricks Notebook 인증은 클러스터의 드라이버 노드에서만 작동하며 클러스터의 작업자 또는 실행기 노드에서는 작동하지 않습니다.

Databricks Runtime 13.3 LTS 이상은 설치된 Python 0.1.7 이상 버전용 Databricks SDK를 사용하여 기본 Azure Databricks Notebook 인증을 지원합니다. Databricks Runtime 10.4 LTS 이상은 설치된 Python 0.1.10 이상 버전용 Databricks SDK를 사용하여 기본 Azure Databricks Notebook 인증을 지원합니다. 그러나 Databricks Runtime 버전에 관계없이 기본 Azure Databricks Notebook 인증과의 최대 호환성을 위해 Python 0.6.0 이상 버전용 Databricks SDK를 설치하거나 업그레이드하는 것이 좋습니다.

Azure Databricks 계정 수준 API를 호출하거나 다음과 같이 기본 Azure Databricks Notebook 인증 이외의 Azure Databricks 인증 유형을 사용하려는 경우 Azure Databricks 클러스터에 Python용 Databricks SDK를 설치하거나 업그레이드해야 합니다.

인증 유형 Python용 Databricks SDK 버전
OAuth M2M(machine-to-machine) 인증 0.18.0 이상
OAuth 사용자 대 컴퓨터(U2M) 인증 0.19.0 이상
Microsoft Entra ID 보안 주체 인증 모든 버전
Azure CLI 인증 모든 버전
Databricks 개인용 액세스 토큰 인증 모든 버전

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

Azure Databricks Notebook 인증은 Azure Databricks 구성 프로필에서 작동하지 않습니다.

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.widgetsdbutils.jobs 명령 그룹에만 액세스할 수 있습니다.
  • Azure Databricks 클러스터에 연결된 Azure Databricks Notebook에서 Databricks Utilities는 dbutils.fs, dbutils.secretsdbutils.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 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 카탈로그 볼륨에 액세스하기 위해 WorkspaceClientfiles 기능에 대한 다양한 호출을 보여 줍니다.

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 모의 라이브러리를 사용합니다.

예를 들어, 다음과 같이 새 클러스터에 대한 정보를 반환하는 create_cluster 함수를 포함하고 있으며 helpers.py(으)로 명명된 파일이 있다고 가정합니다.

# 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

다음과 같이 create_cluster 함수를 호출하며 main.py(으)로 명명된 파일이 제공됩니다.

# 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 ==========================

추가 리소스

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