Соединитель Databricks SQL для Python

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

Требования

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

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

• Установите соединитель SQL Databricks для Python на компьютере разработки, выполнив или python -m pip install databricks-sql-connector.pip install databricks-sql-connector

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

  Кластер

  • Имя узла сервера кластера. Его можно получить из значения Server Hostname (Имя узла сервера) на вкладке Advanced Options (Дополнительные параметры) > JDBC/ODBC для кластера.
  • Путь HTTP кластера. Его можно получить из значения HTTP Path (Путь HTTP) на вкладке Advanced Options (Дополнительные параметры) > JDBC/ODBC для кластера.

  Хранилище SQL

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

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

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

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

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

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

Проверка подлинности маркера личного доступа Databricks

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

1. В рабочей области Azure Databricks щелкните имя пользователя Azure Databricks в верхней строке и выберите "Параметры " в раскрывающемся списке.
2. Щелкните "Разработчик".
3. Рядом с маркерами доступа нажмите кнопку "Управление".
4. Щелкните Generate new token (Создание нового маркера).
5. (Необязательно) Введите комментарий, который поможет определить этот маркер в будущем и изменить время существования маркера по умолчанию в течение 90 дней. Чтобы создать маркер без времени существования (не рекомендуется), оставьте поле время существования (дни) пустым (пустым).
6. Щелкните Создать.
7. Скопируйте отображаемый маркер в безопасное расположение и нажмите кнопку "Готово".

Чтобы проверить подлинность соединителя 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:
# ...

Проверка подлинности на компьютере (M2M) OAuth

Соединитель 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 M2M)". Запишите значение 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, необходимо предоставить соединитель SQL Databricks для Python с маркером идентификатора Microsoft Entra ID. Чтобы создать маркер доступа идентификатора Microsoft Entra, сделайте следующее:

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

Маркеры идентификатора Microsoft Entra имеют время существования по умолчанию около 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:
# ...

Примеры

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

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

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

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

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

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

В следующем примере кода показано, как вызвать соединитель 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

Соединитель Databricks SQL позволяет записывать локальные файлы в тома каталога 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'"
    )

Настройка журнала

Соединитель Databricks SQL использует стандартный модуль ведения журнала 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 таблицы в samples схеме каталога nyctaxi :

# 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 таблицы, макетирование не является абсолютно обязательным в этом примере. Однако макетирование позволяет быстро выполнять тесты, не ожидая фактического подключения к рабочей области. Кроме того, макетирование позволяет выполнять имитированные тесты несколько раз для функций, которые могут изменить состояние таблицы, например 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.

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

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

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

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

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