Коннектор Databricks SQL для Python

Соединитель SQL Databricks для Python — это библиотека Python, которая позволяет использовать код Python для выполнения команд SQL в кластерах Azure Databricks и хранилищах SQL Databricks. Соединитель SQL Databricks для Python проще в настройке и использовании, чем аналогичные библиотеки Python, такие как pyodbc. Эта библиотека соответствует PEP 249 — спецификации API баз данных Python версии 2.0.

Соединитель SQL Databricks для Python также поддерживает диалект SQLAlchemy для Azure Databricks, но его необходимо установить для использования этих функций. См. статью Об использовании SQLAlchemy с Azure Databricks.

Требования

• Компьютер разработки под управлением Python >=3.8 и <=3.11.
• Databricks рекомендует использовать виртуальные среды Python, такие как venv, которые включены в Python. Виртуальные среды помогают обеспечить использование правильных версий Python и Соединителя SQL Databricks для Python вместе. Настройка и использование виртуальных сред выходят за рамки этой статьи. Дополнительные сведения см. в статье "Создание виртуальных сред".
• Существующий кластер или хранилище SQL.

Начало работы

• Установите коннектор SQL Databricks для Python. PyArrow является необязательной зависимостью соединителя SQL Databricks для Python и не устанавливается по умолчанию в версии 4.0.0 и выше соединителя. Если PyArrow не установлен, такие функции, как CloudFetch, и другие возможности Apache Arrow недоступны, что может повлиять на производительность при обработке больших объемов данных.

  • Чтобы установить бережливый соединитель, используйте pip install databricks-sql-connector.
  • Чтобы установить полный соединитель, включая PyArrow, используйте pip install databricks-sql-connector[pyarrow].

• Соберите следующие сведения о кластере или хранилище SQL, которое вы хотите использовать:

  Кластер

  • Имя узла сервера кластера. Вы можете получить это из значения имени узла сервера на вкладке Дополнительные параметры JDBC/ODBC для вашего кластера.
  • Путь HTTP кластера. Вы можете получить это из значения HTTP Path на вкладке «Дополнительные параметры» > JDBC/ODBC для вашего кластера.

  Хранилище SQL

  • Имя узла сервера хранилища SQL. Вы можете получить это из значения имени узла сервера на вкладке "Сведения о подключении" для вашего хранилища SQL.
  • Путь HTTP хранилища SQL. Вы можете получить это из значения HTTP-пути на вкладке Сведения о подключении для вашего SQL-хранилища.

Проверка подлинности

Соединитель SQL Databricks для Python поддерживает следующие типы проверки подлинности Azure Databricks:

• Аутентификация с использованием личного токена доступа Databricks
• Аутентификация токена Microsoft Entra ID
• Аутентификация OAuth для машинного взаимодействия (M2M)
• Проверка подлинности OAuth между пользователем и машиной (U2M)

Соединитель SQL Databricks для Python пока не поддерживает следующие типы проверки подлинности Azure Databricks:

• Проверка подлинности управляемых удостоверений Azure
• Проверка подлинности принципала службы MS Entra
• Проверка подлинности Azure CLI

Аутентификация с использованием личного токена доступа Databricks

Чтобы использовать коннектор Databricks SQL для Python с аутентификацией с помощью личного маркера доступа Azure Databricks, необходимо сначала создать личный маркер доступа Azure Databricks. Для этого выполните действия из раздела маркер персонального доступа Azure Databricks для пользователей рабочей области.

Чтобы аутентифицировать соединитель SQL Databricks для Python, используйте следующий фрагмент кода. В этом фрагменте предполагается, что вы установили следующие переменные среды:

• Установите DATABRICKS_SERVER_HOSTNAME значение имени узла сервера для вашего кластера или SQL-хранилища.
• Установите для DATABRICKS_HTTP_PATH значение пути HTTP для вашего кластера или SQL-хранилища.
• DATABRICKS_TOKEN должен быть установлен в качестве персонального токена доступа Azure Databricks.

Чтобы задать переменные среды, ознакомьтесь с документацией операционной системы.

from databricks import sql
import os

with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                 access_token    = os.getenv("DATABRICKS_TOKEN")) as connection:
# ...

Аутентификация OAuth между машинами (M2M)

Соединитель SQL Databricks для Python версии 2.7.0 и выше поддерживает проверку подлинности OAuth на компьютере (M2M). Необходимо также установить пакет SDK Databricks для Python 0.18.0 или более поздней версии (например, выполнив pip install databricks-sdk или python -m pip install databricks-sdk).

Чтобы использовать соединитель SQL Databricks для Python с проверкой подлинности OAuth M2M, необходимо выполнить следующие действия.

1. Создайте сервисный принципал Azure Databricks в рабочей области Azure Databricks и создайте секрет OAuth для этого сервисного принципала.

   Сведения о создании принципала услуг и его секрета OAuth см. в статье Авторизация автоматического доступа к ресурсам Azure Databricks, используя принципал услуг и OAuth. Запишите значение UUID или идентификатора приложения принципала службы и значение секрета для OAuth-ключа принципала службы.

2. Предоставьте служебному участнику доступ к вашему кластеру или хранилищу.

   Чтобы предоставить учетной записи службы доступ к вашему кластеру или хранилищу, ознакомьтесь с разрешениями на вычисления или управлением хранилищем SQL.

Чтобы аутентифицировать соединитель SQL Databricks для Python, используйте следующий фрагмент кода. В этом фрагменте предполагается, что вы установили следующие переменные среды:

• Значение параметра DATABRICKS_SERVER_HOSTNAME установлено в значение имени узла сервера для вашего кластера или хранилища SQL.
• Установите для DATABRICKS_HTTP_PATH значение пути HTTP для вашего кластера или SQL-хранилища.
• DATABRICKS_CLIENT_IDУстановите для главного объекта службы значение UUID или идентификатор приложения.
• DATABRICKS_CLIENT_SECRET, установите значение секрет для OAuth секрета служебного принципала.

Чтобы задать переменные среды, ознакомьтесь с документацией операционной системы.

from databricks.sdk.core import Config, oauth_service_principal
from databricks import sql
import os

server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME")

def credential_provider():
  config = Config(
    host          = f"https://{server_hostname}",
    client_id     = os.getenv("DATABRICKS_CLIENT_ID"),
    client_secret = os.getenv("DATABRICKS_CLIENT_SECRET"))
  return oauth_service_principal(config)

with sql.connect(server_hostname      = server_hostname,
                 http_path            = os.getenv("DATABRICKS_HTTP_PATH"),
                 credentials_provider = credential_provider) as connection:
# ...

Проверка подлинности токена идентификатора Microsoft Entra

Чтобы использовать соединитель Databricks SQL для Python с аутентификацией токеном Microsoft Entra ID, необходимо предоставить этому соединителю токен Microsoft Entra ID. Чтобы создать токен доступа Microsoft Entra ID, сделайте следующее:

• Для пользователя Azure Databricks можно использовать Azure CLI. См. получение токенов Microsoft Entra ID для пользователей с помощью Azure CLI.
• Сведения о служебном принципале Microsoft Entra ID см. в статье Получение токена доступа Microsoft Entra ID с помощью Azure CLI. Чтобы создать управляемый принципал службы Microsoft Entra ID, см. раздел Субъекты-службы.

Токены Microsoft Entra ID имеют срок действия по умолчанию около 1 часа. Чтобы создать новый маркер идентификатора Microsoft Entra, повторите этот процесс.

Чтобы аутентифицировать соединитель SQL Databricks для Python, используйте следующий фрагмент кода. В этом фрагменте предполагается, что вы установили следующие переменные среды:

• Задайте DATABRICKS_SERVER_HOSTNAME значение имени узла сервера для кластера или хранилища SQL.
• Установите значение DATABRICKS_HTTP_PATH для пути HTTP вашего кластера или склада SQL.
• Установите DATABRICKS_TOKEN в качестве токена идентификатора Microsoft Entra ID.

Чтобы задать переменные среды, ознакомьтесь с документацией операционной системы.

from databricks import sql
import os

with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                 access_token    = os.getenv("DATABRICKS_TOKEN")) as connection:
# ...

Аутентификация пользователя на машине по OAuth (U2M)

Соединитель SQL Databricks для Python версии 2.7.0 и более поздних версий поддерживает проверку подлинности OAuth на компьютере (U2M). Кроме того, необходимо установить пакет SDK Databricks для Python 0.19.0 или более поздней версии (например, запустив pip install databricks-sdk или python -m pip install databricks-sdk).

Чтобы аутентифицировать коннектор SQL Databricks для Python с помощью аутентификации OAuth U2M, используйте следующий фрагмент кода. Проверка подлинности OAuth U2M использует вход в режиме реального времени и согласие на проверку подлинности целевой учетной записи пользователя Azure Databricks. В этом фрагменте предполагается, что вы установили следующие переменные среды:

• Задайте DATABRICKS_SERVER_HOSTNAME значение имени узла сервера для кластера или хранилища SQL.
• Установите значение DATABRICKS_HTTP_PATH для пути HTTP вашего кластера или склада SQL.

Чтобы задать переменные среды, ознакомьтесь с документацией операционной системы.

from databricks import sql
import os

with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                 auth_type       = "databricks-oauth") as connection:
# ...

Примеры

В следующих примерах кода показано, как использовать соединитель SQL Databricks для Python для запроса и вставки данных, метаданных запроса, управления курсорами и подключениями и настройки ведения журнала.

В этих примерах кода значения переменных подключения server_hostname, http_pathи access_token извлекаются из соответствующих переменных среды.

• DATABRICKS_SERVER_HOSTNAME, которая предоставляет значение Имени узла сервера в соответствии с требованиями.
• DATABRICKS_HTTP_PATH, который представляет значение Пути HTTP в соответствии с требованиями.
• DATABRICKS_TOKEN, которая представляет ваш токен доступа из требований.

Вы можете использовать другие подходы к получению этих значений переменных подключения. Использование переменных среды — это лишь один из многих подходов.

• Установить User-Agent
• Запрос данных
• Вставка данных
• Метаданные запроса
• Управление курсорами и подключениями
• Управление файлами в томах каталога Unity
• Настройка логирования

Комплект User-Agent

Следующий пример кода демонстрирует, как настроить приложение User-Agent product_name для мониторинга использования.

from databricks import sql
import os

with sql.connect(server_hostname   = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path         = os.getenv("DATABRICKS_HTTP_PATH"),
                 access_token      = os.getenv("DATABRICKS_TOKEN"),
                 user_agent_entry = "product_name") as connection:
  with connection.cursor() as cursor:
    cursor.execute("SELECT 1 + 1")
    result = cursor.fetchall()

    for row in result:
      print(row)

Запрос данных

В следующем примере кода показано, как вызвать соединитель SQL Databricks для Python для выполнения базовой команды SQL в кластере или хранилище SQL. Эта команда возвращает первые две строки из таблицы trips в схеме samples каталога nyctaxi.

from databricks import sql
import os

with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                 access_token    = os.getenv("DATABRICKS_TOKEN")) as connection:

  with connection.cursor() as cursor:
    cursor.execute("SELECT * FROM samples.nyctaxi.trips LIMIT 2")
    result = cursor.fetchall()

    for row in result:
      print(row)

Вставка данных

В следующем примере показано, как вставлять небольшие объемы данных (тысячи строк):

from databricks import sql
import os

with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                 access_token    = os.getenv("DATABRICKS_TOKEN")) as connection:

  with connection.cursor() as cursor:
    cursor.execute("CREATE TABLE IF NOT EXISTS squares (x int, x_squared int)")

    squares = [(i, i * i) for i in range(100)]
    values = ",".join([f"({x}, {y})" for (x, y) in squares])

    cursor.execute(f"INSERT INTO squares VALUES {values}")

    cursor.execute("SELECT * FROM squares LIMIT 10")

    result = cursor.fetchall()

    for row in result:
      print(row)

Для больших объемов данных необходимо сначала передать данные в облачное хранилище, а затем выполнить команду COPY INTO.

Метаданные запроса

Есть специальные методы для извлечения метаданных. В следующем примере извлекаются метаданные о столбцах в примере таблицы:

from databricks import sql
import os

with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                 access_token    = os.getenv("DATABRICKS_TOKEN")) as connection:

  with connection.cursor() as cursor:
    cursor.columns(schema_name="default", table_name="squares")
    print(cursor.fetchall())

Управление курсорами и подключениями

Рекомендуется закрыть все подключения и курсоры, которые больше не используются. Это позволит освободить ресурсы в кластерах Azure Databricks и хранилищах SQL Databricks.

Вы можете применить диспетчер контекста (синтаксис with, используемый в предыдущих примерах) для управления ресурсами или явно вызватьclose:

from databricks import sql
import os

connection = sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                         http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                         access_token    = os.getenv("DATABRICKS_TOKEN"))

cursor = connection.cursor()

cursor.execute("SELECT * from range(10)")
print(cursor.fetchall())

cursor.close()
connection.close()

Управление файлами в томах каталога Unity

Соединитель SQL Databricks позволяет записывать локальные файлы в каталог Unity тома, загружать файлы из томов и удалять файлы из томов, как показано в следующем примере:

from databricks import sql
import os

# For writing local files to volumes and downloading files from volumes,
# you must set the staging_allows_local_path argument to the path to the
# local folder that contains the files to be written or downloaded.
# For deleting files in volumes, you must also specify the
# staging_allows_local_path argument, but its value is ignored,
# so in that case its value can be set for example to an empty string.
with sql.connect(server_hostname            = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                 http_path                  = os.getenv("DATABRICKS_HTTP_PATH"),
                 access_token               = os.getenv("DATABRICKS_TOKEN"),
                 staging_allowed_local_path = "/tmp/") as connection:

  with connection.cursor() as cursor:

    # Write a local file to the specified path in a volume.
    # Specify OVERWRITE to overwrite any existing file in that path.
    cursor.execute(
      "PUT '/temp/my-data.csv' INTO '/Volumes/main/default/my-volume/my-data.csv' OVERWRITE"
    )

    # Download a file from the specified path in a volume.
    cursor.execute(
      "GET '/Volumes/main/default/my-volume/my-data.csv' TO '/tmp/my-downloaded-data.csv'"
    )

    # Delete a file from the specified path in a volume.
    cursor.execute(
      "REMOVE '/Volumes/main/default/my-volume/my-data.csv'"
    )

Конфигурировать журнал

Соединитель SQL Databricks использует стандартный модуль ведения журнала Python. Уровень ведения журнала можно настроить следующим образом:

from databricks import sql
import os, logging

logging.getLogger("databricks.sql").setLevel(logging.DEBUG)
logging.basicConfig(filename = "results.log",
                    level    = logging.DEBUG)

connection = sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
                         http_path       = os.getenv("DATABRICKS_HTTP_PATH"),
                         access_token    = os.getenv("DATABRICKS_TOKEN"))

cursor = connection.cursor()

cursor.execute("SELECT * from range(10)")

result = cursor.fetchall()

for row in result:
   logging.debug(row)

cursor.close()
connection.close()

Тестирование

Чтобы протестировать код, используйте платформы тестов Python, такие как pytest. Чтобы протестировать код в имитированных условиях без вызова конечных точек REST API Azure Databricks или изменения состояния учетных записей Или рабочих областей Azure Databricks, можно использовать библиотеки макетирования Python, такие как unittest.mock.

Например, учитывая следующий файл с именем helpers.py, содержащий функцию get_connection_personal_access_token, которая использует личный маркер доступа Azure Databricks для возврата подключения к рабочей области Azure Databricks, а также функцию select_nyctaxi_trips, использующую это подключение для получения указанного количества строк данных из таблицы trips в схеме nyctaxi каталога samples:

# helpers.py

from databricks import sql
from databricks.sql.client import Connection, List, Row, Cursor

def get_connection_personal_access_token(
  server_hostname: str,
  http_path: str,
  access_token: str
) -> Connection:
  return sql.connect(
    server_hostname = server_hostname,
    http_path = http_path,
    access_token = access_token
  )

def select_nyctaxi_trips(
  connection: Connection,
  num_rows: int
) -> List[Row]:
  cursor: Cursor = connection.cursor()
  cursor.execute(f"SELECT * FROM samples.nyctaxi.trips LIMIT {num_rows}")
  result: List[Row] = cursor.fetchall()
  return result

И учитывая следующий файл с именем main.py, который вызывает функции get_connection_personal_access_token и select_nyctaxi_trips.

# main.py

from databricks.sql.client import Connection, List, Row
import os
from helpers import get_connection_personal_access_token, select_nyctaxi_trips

connection: Connection = get_connection_personal_access_token(
  server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
  http_path = os.getenv("DATABRICKS_HTTP_PATH"),
  access_token = os.getenv("DATABRICKS_TOKEN")
)

rows: List[Row] = select_nyctaxi_trips(
  connection = connection,
  num_rows = 2
)

for row in rows:
  print(row)

Следующий файл с именем test_helpers.py проверяет, возвращает ли select_nyctaxi_trips функция ожидаемый ответ. Вместо создания реального подключения к целевой рабочей области этот тест макетирует Connection объект. Тест также макетирует некоторые данные, соответствующие схеме и значениям, которые находятся в реальных данных. Тест возвращает мокаемые данные через мокаемое подключение, а затем проверяет, соответствует ли одно из значений строк мокаемых данных ожидаемому значению.

# test_helpers.py

import pytest
from databricks.sql.client import Connection, List, Row
from datetime import datetime
from helpers import select_nyctaxi_trips
from unittest.mock import create_autospec

@pytest.fixture
def mock_data() -> List[Row]:
  return [
    Row(
      tpep_pickup_datetime = datetime(2016, 2, 14, 16, 52, 13),
      tpep_dropoff_datetime = datetime(2016, 2, 14, 17, 16, 4),
      trip_distance = 4.94,
      fare_amount = 19.0,
      pickup_zip = 10282,
      dropoff_zip = 10171
    ),
    Row(
      tpep_pickup_datetime = datetime(2016, 2, 4, 18, 44, 19),
      tpep_dropoff_datetime = datetime(2016, 2, 4, 18, 46),
      trip_distance = 0.28,
      fare_amount = 3.5,
      pickup_zip = 10110,
      dropoff_zip = 10110
    )
  ]

def test_select_nyctaxi_trips(mock_data: List[Row]):
  # Create a mock Connection.
  mock_connection = create_autospec(Connection)

  # Set the mock Connection's cursor().fetchall() to the mock data.
  mock_connection.cursor().fetchall.return_value = mock_data

  # Call the real function with the mock Connection.
  response: List[Row] = select_nyctaxi_trips(
    connection = mock_connection,
    num_rows = 2)

  # Check the value of one of the mocked data row's columns.
  assert response[1].fare_amount == 3.5

Так как функция select_nyctaxi_trips содержит инструкцию SELECT и поэтому не изменяет состояние таблицы trips, использование mock-объектов не является абсолютно необходимым в этом примере. Однако мокирование позволяет быстро проводить ваши тесты, не ожидая реального подключения к рабочей области. Кроме того, макетирование позволяет выполнять имитированные тесты несколько раз для функций, которые могут изменить состояние таблицы, например INSERT INTO, UPDATEи DELETE FROM.

Справочник по API

• Пакет
• Модуль
• Классы
• ConnectionКласс
• CursorКласс
• RowКласс
• Преобразования типов

Пакет

databricks-sql-connector

Использование: pip install databricks-sql-connector

См. также databricks-sql-connector в индексе пакета Python (PyPI).

Модуль

databricks.sql

Использование: from databricks import sql

Классы

К выбранным классам относятся следующие:

Класс Connection

Чтобы создать объект Connection, вызовите метод databricks.sql.connect со следующими параметрами:

Выбранные Connection методы включают следующие:

Класс Cursor

Чтобы создать Cursor объект, вызовите Connection метод класса cursor .

К выбранным Cursor атрибутам относятся следующие:

Выбранные Cursor методы включают следующие:

Класс Row

Класс row — это структура данных, похожая на кортеж, которая представляет отдельную строку результата. Если строка содержит столбец с именем "my_column", вы можете получить доступ к полю "my_column"row через row.my_column. Кроме того, для доступа к полям вы можете использовать числовые индексы, например row[0]. Если имя столбца не допускается в качестве имени метода атрибута (например, начинается с цифры), вы можете получить доступ к полю как row["1_my_column"].

Начиная с версии 1.0

К выбранным Row методам относятся:

| asDict

Возвращает представление словаря строки, которая индексируется по именам полей. Если имена полей дублируются, одно из них (но только одно) будет возвращено в словаре. Возвращаемое дублирующееся поле не определяется.

Нет параметров.

Возвращает dict список полей. |

Преобразование типов

В следующей таблице сопоставляется типы данных Apache Spark SQL с эквивалентами их типов данных Python.

Устранение неполадок

Сообщение tokenAuthWrapperInvalidAccessToken: Invalid access token

Проблема. При выполнении кода отображается примерно такое сообщение: Error during request to server: tokenAuthWrapperInvalidAccessToken: Invalid access token.

Возможная причина. Значение, переданное в access_token, не является допустимым личным маркером доступа Azure Databricks.

Рекомендуемое исправление. Проверьте правильность значения, переданного в access_token, и повторите попытку.

Сообщение gaierror(8, 'nodename nor servname provided, or not known')

Проблема. При выполнении кода отображается примерно такое сообщение: Error during request to server: gaierror(8, 'nodename nor servname provided, or not known').

Возможная причина. Значение, передаваемое в server_hostname, не является правильным именем узла.

Рекомендуемое исправление. Проверьте правильность значения, переданного в server_hostname, и повторите попытку.

Дополнительные сведения о поиске имени узла сервера см. в статье Получение сведений о подключении для вычислительного ресурса Azure Databricks.

Сообщение IpAclError

Проблема: При выполнении кода отображается сообщение Error during request to server: IpAclValidation, когда вы пытаетесь использовать коннектор в записной книжке Azure Databricks.

Возможная причина: возможно, для рабочей области Azure Databricks включен доступ на основе списка разрешенных IP-адресов. При наличии списка разрешенных IP-адресов подключения из кластеров Spark обратно к плоскости управления по умолчанию запрещены.

рекомендуемое исправление. Попросите администратора добавить подсеть уровня вычислений в список разрешенных IP-адресов.

Дополнительные ресурсы

Дополнительные сведения см. в разделе:

• Соединитель SQL Databricks для Python на сайте GitHub
• Типы данных
• Встроенные типы (для bool, bytearray, float, int и str) на веб-сайте Python.
• Дата и время (для datetime.date и datatime.datetime) на веб-сайте Python.
• Десятичное число (для decimal.Decimal) на веб-сайте Python.
• Встроенные константы (для NoneType) на веб-сайте Python.