Databricks SDK pro Python

V tomto článku se dozvíte, jak automatizovat operace Azure Databricks a zrychlit vývoj pomocí sady Databricks SDK pro Python. Tento článek doplňuje dokumentaci k sadě Databricks SDK pro Python na webu Read The Docs a příklady kódu v úložišti Databricks SDK pro Python na GitHubu.

Než začnete

Sadu Databricks SDK pro Python můžete použít z poznámkového bloku Azure Databricks nebo z místního vývojového počítače.

• Pokud chcete použít sadu Databricks SDK pro Python z poznámkového bloku Azure Databricks, přeskočte k použití sady Databricks SDK pro Python z poznámkového bloku Azure Databricks.
• Pokud chcete použít sadu Databricks SDK pro Python z místního vývojového počítače, proveďte kroky v této části.

Než začnete používat sadu Databricks SDK pro Python, váš vývojový počítač musí mít:

• Nakonfigurované ověřování Azure Databricks
• Nainstalovaný Python 3.8 nebo novější Pro automatizaci výpočetních prostředků Azure Databricks doporučuje Databricks, abyste měli nainstalované hlavní a podverze Pythonu, které odpovídají verzi nainstalované na cílovém výpočetním prostředku Azure Databricks. Příklady tohoto článku se spoléhají na automatizaci clusterů s modulem Databricks Runtime 13.3 LTS, který má nainstalovaný Python 3.10. Správnou verzi najdete v poznámkách k verzi databricks Runtime a kompatibilitu pro verzi Databricks Runtime vašeho clusteru.
• Databricks doporučuje vytvořit a aktivovat virtuální prostředí Pythonu pro každý projekt Pythonu, který používáte se sadou Databricks SDK pro Python. Virtuální prostředí Pythonu pomáhají zajistit, aby váš projekt kódu používal kompatibilní verze balíčků Pythonu a Pythonu (v tomto případě balíček Databricks SDK pro Python). Další informace o virtuálních prostředích Pythonu najdete v tématu venv nebo Básně.

Začínáme se sadou Databricks SDK pro Python

Tato část popisuje, jak začít se sadou Databricks SDK pro Python z místního vývojového počítače. Pokud chcete použít sadu Databricks SDK pro Python z poznámkového bloku Azure Databricks, přeskočte k použití sady Databricks SDK pro Python z poznámkového bloku Azure Databricks.

1. Na vývojovém počítači s nakonfigurovaným ověřováním Azure Databricks už je nainstalovaný Python a vaše virtuální prostředí Pythonu už aktivované, nainstalujte balíček databricks-sdk (a jeho závislosti) z indexu balíčků Pythonu (PyPI), a to následujícím způsobem:

   Venv

   Slouží pip k instalaci databricks-sdk balíčku. (V některých systémech možná budete muset nahradit pip3 , pipzde a v celém.)

   pip3 install databricks-sdk
   

   Poezie

   poetry add databricks-sdk
   

   Pokud chcete nainstalovat konkrétní verzi databricks-sdk balíčku, když je sada Databricks SDK pro Python v beta verzi, podívejte se do historie verzí balíčku. Například pro instalaci verze 0.1.6:

   Venv

   pip3 install databricks-sdk==0.1.6
   

   Poezie

   poetry add databricks-sdk==0.1.6
   

   Pokud chcete upgradovat existující instalaci balíčku Databricks SDK pro Python na nejnovější verzi, spusťte následující příkaz:

   Venv

   pip3 install --upgrade databricks-sdk
   

   Poezie

   poetry add databricks-sdk@latest
   

   Pokud chcete zobrazit aktuální Version a další podrobnosti balíčku Databricks SDK pro Python, spusťte následující příkaz:

   Venv

   pip3 show databricks-sdk
   

   Poezie

   poetry show databricks-sdk
   

2. Ve virtuálním prostředí Pythonu vytvořte soubor kódu Pythonu, který naimportuje sadu Databricks SDK pro Python. Následující příklad v souboru s názvem main.py s následujícím obsahem jednoduše vypíše všechny clustery v pracovním prostoru Azure Databricks:

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

3. Spusťte soubor kódu Pythonu za předpokladu, že spustíte python příkaz s názvem main.pysouboru:

   Venv

   python3.10 main.py
   

   Poezie

   Pokud jste v prostředí virtuálního prostředí:

   python3.10 main.py
   

   Pokud nejste v prostředí virtuálního prostředí:

   poetry run python3.10 main.py
   

   Poznámka:

   Když v předchozím volání w = WorkspaceClient()nenastavíte žádné argumenty, použije sada Databricks SDK pro Python výchozí proces pro pokus o ověření Azure Databricks. Pokud chcete toto výchozí chování přepsat, přečtěte si následující část ověřování .

Ověření sady Databricks SDK pro Python pomocí účtu nebo pracovního prostoru Azure Databricks

Tato část popisuje, jak ověřit sadu Databricks SDK pro Python z místního vývojového počítače do účtu nebo pracovního prostoru Azure Databricks. Pokud chcete ověřit sadu Databricks SDK pro Python z poznámkového bloku Azure Databricks, přeskočte k použití sady Databricks SDK pro Python z poznámkového bloku Azure Databricks.

Sada Databricks SDK pro Python implementuje standard sjednoceného ověřování klienta Databricks, konsolidovaný a konzistentní přístup architektury a programového přístupu k ověřování. Tento přístup pomáhá zajistit centralizovanější a předvídatelnější nastavení a automatizaci ověřování pomocí Azure Databricks. Umožňuje konfigurovat ověřování Databricks jednou a pak tuto konfiguraci používat napříč několika nástroji a sadami SDK Databricks bez dalších změn konfigurace ověřování. Další informace, včetně kompletních příkladů kódu v Pythonu, najdete v tématu Jednotné ověřování klienta Databricks.

Mezi dostupné vzory kódování pro inicializaci ověřování Databricks pomocí sady Databricks SDK pro Python patří:

• Výchozí ověřování Databricks můžete použít jedním z následujících způsobů:

  • Vytvořte nebo identifikujte vlastní konfigurační profil Databricks s požadovanými poli pro cílový typ ověřování Databricks. Potom nastavte proměnnou DATABRICKS_CONFIG_PROFILE prostředí na název vlastního konfiguračního profilu.
  • Nastavte požadované proměnné prostředí pro cílový typ ověřování Databricks.

  Pak vytvořte instanci objektu WorkspaceClient s výchozím ověřováním Databricks následujícím způsobem:

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

• Požadovaná pole se pevně zakódují, ale nedoporučuje se, protože riskuje zveřejnění citlivých informací v kódu, jako jsou osobní přístupové tokeny Azure Databricks. Následující příklad pevně kóduje hodnoty hostitele Azure Databricks a přístupového tokenu pro ověřování tokenů Databricks:

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

Viz také ověřování v dokumentaci k sadě Databricks SDK pro Python.

Použití sady Databricks SDK pro Python z poznámkového bloku Azure Databricks

Sadu Databricks SDK pro Python můžete volat z poznámkového bloku Azure Databricks s připojeným clusterem Azure Databricks s nainstalovanou sadou Databricks SDK pro Python. Ve výchozím nastavení se instaluje ve všech clusterech Azure Databricks, které používají Databricks Runtime 13.3 LTS nebo vyšší. V případě clusterů Azure Databricks, které používají Databricks Runtime 12.2 LTS a níže, musíte nejprve nainstalovat sadu Databricks SDK pro Python. Viz krok 1: Instalace nebo upgrade sady Databricks SDK pro Python.

Pokud chcete zobrazit sadu Databricks SDK pro Python, která je nainstalovaná pro konkrétní verzi databricks Runtime, přečtěte si část Nainstalované knihovny Pythonu v poznámkách k verzi Databricks Runtime pro danou verzi.

Databricks doporučuje nainstalovat nejnovější dostupnou verzi sady SDK z PiPy, ale minimálně nainstalovat nebo upgradovat na Sadu Databricks SDK pro Python 0.6.0 nebo novější, protože výchozí ověřování poznámkového bloku Azure Databricks používá verze 0.6.0 a vyšší ve všech verzích Databricks Runtime.

Následující tabulka popisuje podporu ověřování poznámkových bloků pro sadu Databricks SDK pro verze Pythonu a Databricks Runtime:

Výchozí ověřování poznámkového bloku Azure Databricks závisí na dočasném tokenu pat Azure Databricks, který Azure Databricks automaticky generuje na pozadí pro vlastní použití. Azure Databricks odstraní tento dočasný token, jakmile poznámkový blok přestane běžet.

Pokud chcete volat rozhraní API na úrovni účtu Azure Databricks nebo pokud chcete použít jiný typ ověřování Databricks než výchozí ověřování poznámkového bloku Databricks, podporují se také následující typy ověřování:

Ověřování spravovaných identit Azure se zatím nepodporuje.

Krok 1: Instalace nebo upgrade sady Databricks SDK pro Python

1. Poznámkové bloky Pythonu v Azure Databricks můžou používat sadu Databricks SDK pro Python stejně jako jakoukoli jinou knihovnu Pythonu. Pokud chcete nainstalovat nebo upgradovat knihovnu Databricks SDK pro Python v připojeném clusteru Azure Databricks, spusťte %pip příkaz magic z buňky poznámkového bloku následujícím způsobem:

   %pip install databricks-sdk --upgrade
   

2. Po spuštění %pip příkazu magic musíte restartovat Python, aby byla nainstalovaná nebo upgradovaná knihovna dostupná pro poznámkový blok. Uděláte to tak, že z buňky poznámkového bloku hned za buňkou %pip spustíte následující příkaz:

   dbutils.library.restartPython()
   

3. Pokud chcete zobrazit nainstalovanou verzi sady Databricks SDK pro Python, spusťte z buňky poznámkového bloku následující příkaz:

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

Krok 2: Spuštění kódu

Vbuňkách Následující příklad používá výchozí ověřování poznámkového bloku Azure Databricks k výpisu všech clusterů v pracovním prostoru Azure Databricks:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

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

Když spustíte tuto buňku, zobrazí se seznam názvů všech dostupných clusterů v pracovním prostoru Azure Databricks.

Pokud chcete použít jiný typ ověřování Azure Databricks, podívejte se na metody ověřování Azure Databricks a kliknutím na odpovídající odkaz zobrazíte další technické podrobnosti.

Použití nástrojů Databricks

Odkazy na nástroje Databricks (dbutils) můžete volat ze sady Databricks SDK pro Python spuštěný na místním vývojovém počítači nebo v poznámkovém bloku Azure Databricks.

• Z místního vývojového počítače mají nástroje Databricks přístup pouze ke skupinám dbutils.fspříkazů , dbutils.secrets, dbutils.widgetsa dbutils.jobs příkazů.
• Z poznámkového bloku Azure Databricks, který je připojený ke clusteru Azure Databricks, má nástroje Databricks přístup ke všem dostupným skupinám příkazů Databricks Utilities, nejen dbutils.fsdbutils.secretsk a dbutils.widgets. Kromě toho dbutils.notebook je skupina příkazů omezena pouze na dvě úrovně příkazů, například dbutils.notebook.run nebo dbutils.notebook.exit.

Pokud chcete volat nástroje Databricks z místního vývojového počítače nebo poznámkového bloku Azure Databricks, použijte dbutils ho v rámci WorkspaceClient. Tento příklad kódu používá výchozí ověřování poznámkového bloku Azure Databricks k volání dbutils do seznamu WorkspaceClient cest všech objektů v kořenovém adresáři DBFS pracovního prostoru.

from databricks.sdk import WorkspaceClient

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

for f in d:
  print(f.path)

Případně můžete volat dbutils přímo. Jste ale omezeni na použití výchozího ověřování poznámkového bloku Azure Databricks. Tento příklad kódu volá dbutils přímo seznam všech objektů v kořenovém adresáři DBFS pracovního prostoru.

from databricks.sdk.runtime import *

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

for f in d:
  print(f.path)

Pro přístup ke svazkům katalogu Unity použijte files v souboru WorkspaceClient. Viz Správa souborů ve svazcích katalogu Unity. Pro přístup ke svazkům nemůžete použít dbutils sám ani v rámci WorkspaceClient sebe.

Viz také Interakce s nástroji dbutils.

Příklady kódu

Následující příklady kódu ukazují, jak pomocí sady Databricks SDK pro Python vytvářet a odstraňovat clustery, spouštět úlohy a vypisovat skupiny na úrovni účtu. Tyto příklady kódu používají výchozí ověřování poznámkového bloku Azure Databricks. Podrobnosti o výchozím ověřování poznámkového bloku Azure Databricks najdete v tématu Použití sady Databricks SDK pro Python z poznámkového bloku Azure Databricks. Podrobnosti o výchozím ověřování mimo poznámkové bloky najdete v tématu Ověření sady Databricks SDK pro Python pomocí účtu nebo pracovního prostoru Azure Databricks.

Další příklady kódu najdete v příkladech v úložišti Databricks SDK pro Python na GitHubu. Viz také:

• Referenční informace k rozhraním API pracovního prostoru Databricks

• Referenční informace k rozhraním API účtu Databricks

• Vytvoření clusteru

• Trvalé odstranění clusteru

• Vytvoření úlohy

• Vytvoření úlohy, která používá bezserverové výpočetní prostředky

• Správa souborů ve svazcích katalogu Unity

• Výpis skupin na úrovni účtu

Vytvoření clusteru

Tento příklad kódu vytvoří cluster se zadanou verzí databricks Runtime a typem uzlu clusteru. Tento cluster má jeden pracovní proces a cluster se automaticky ukončí po 15 minutách nečinnosti.

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")

Trvalé odstranění clusteru

Tento příklad kódu trvale odstraní cluster se zadaným ID clusteru z pracovního prostoru.

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)

Vytvoření úlohy

Tento příklad kódu vytvoří úlohu Azure Databricks, která spustí zadaný poznámkový blok v zadaném clusteru. Při spuštění kódu získá cestu existujícího poznámkového bloku, id existujícího clusteru a související nastavení úlohy od uživatele v terminálu.

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")

Vytvoření úlohy, která používá bezserverové výpočetní prostředky

Následující příklad vytvoří úlohu, která používá bezserverové výpočetní prostředky pro úlohy:

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",
   )
  ]
)

Správa souborů ve svazcích katalogu Unity

Tento příklad kódu ukazuje různá volání files funkcí v rámci WorkspaceClient přístupu ke svazku katalogu Unity.

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)

Výpis skupin na úrovni účtu

Tento příklad kódu uvádí zobrazované názvy všech dostupných skupin v rámci účtu Azure Databricks.

from databricks.sdk import AccountClient

a = AccountClient()

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

Testování

K otestování kódu použijte testovací architektury Pythonu, jako je pytest. K otestování kódu za simulovaných podmínek bez volání koncových bodů rozhraní REST API Služby Azure Databricks nebo změny stavu účtů nebo pracovních prostorů Azure Databricks použijte knihovny napodobování Pythonu, jako je unittest.mock.

Následující ukázkový soubor s názvem helpers.py obsahuje create_cluster funkci, která vrací informace o novém clusteru:

# 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

Vzhledem k následujícímu create_cluster souboru, main.py který volá funkci:

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

Následující soubor s názvem test_helpers.py testuje, zda create_cluster funkce vrátí očekávanou odpověď. Místo vytvoření clusteru v cílovém pracovním prostoru tento test napodobí WorkspaceClient objekt, definuje nastavení napodobeného objektu a pak předá napodobený objekt funkci create_cluster . Test pak zkontroluje, jestli funkce vrátí očekávané ID nového napodobeného clusteru.

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

Pokud chcete tento test spustit, spusťte pytest příkaz z kořenového adresáře projektu kódu, který by měl vést k výsledkům testů podobných následujícímu:

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

Další materiály

Další informace naleznete v tématu:

• Dokumentace k sadě Databricks SDK pro Python

• Další příklady kódu

• Referenční informace k rozhraním API pracovního prostoru Databricks

• Referenční informace k rozhraním API účtu Databricks

• Protokolování

• Dlouhotrvající operace

• Stránkované odpovědi

• Jednotné přihlašování pomocí OAuth