将 Python 和 pyodbc 连接到 Azure Databricks

可以通过 ODBC 从本地 Python 代码连接到 Azure Databricks 群集或 SQL 仓库中的数据。 为此，可以使用开源 Python 代码模块 pyodbc。

按照这些说明安装、配置和使用 pyodbc。

有关 pyodbc 的详细信息，请参阅 pyodbc Wiki。

要求

• 一台运行以下操作系统之一的本地开发计算机：
  • macOS
  • Windows
  • 支持 .rpm 或 .deb 文件的 Unix 或 Linux 发行版
• pip。
• 对于 Unix、Linux 或 macOS，请安装 Homebrew。
• 一个 Azure Databricks 群集和/或一个 Databricks SQL 仓库。 有关详细信息，请参阅计算配置参考和连接到 SQL 仓库。

步骤 1：下载、安装和配置软件

在此步骤中下载并安装 Databricks ODBC 驱动程序、unixodbc 包和 pyodbc 模块。 （pyodbc 模块需要 Unix、Linux 和 macOS 上的 unixodbc 包。）还可以配置 ODBC 数据源名称 (DSN)，以使用群集或 SQL 仓库进行身份验证并连接到它们。

1. 下载并安装 Databricks ODBC 驱动程序，并为操作系统配置 ODBC DSN。

2. 对于 Unix、Linux 和 macOS，请安装 unixodbc 包：通过终端使用 Homebrew 运行命令 brew install unixodbc。 有关详细信息，请参阅 Homebrew 网站上的 unixodbc。

3. 安装 pyodbc 模块：在终端或命令提示符下，使用 pip 运行命令 pip install pyodbc。 有关详细信息，请参阅 PyPI 网站上的 pyodbc，以及 pyodbc Wiki 中的安装。

步骤 2：测试你的配置

在此步骤中，我们编写并运行 Python 代码，以使用 Azure Databricks 群集或 Databricks SQL 仓库查询 trips 目录的 samples 架构中的 nyctrips 表并显示结果。

1. 创建包含以下内容的名为 pyodbc-demo.py 的文件。 将 <dsn-name> 替换为你之前创建的 ODBC DSN 的名称，保存该文件，然后使用 Python 解释器运行该文件。

   import pyodbc
   
   # Connect to the Databricks cluster by using the
   # Data Source Name (DSN) that you created earlier.
   conn = pyodbc.connect("DSN=<dsn-name>", autocommit=True)
   
   # Run a SQL query by using the preceding connection.
   cursor = conn.cursor()
   cursor.execute(f"SELECT * FROM samples.nyctaxi.trips")
   
   # Print the rows retrieved from the query.
   for row in cursor.fetchall():
     print(row)
   

2. 要加快代码运行速度，请。

3. 使用 Python 解释器运行 pyodbc-demo.py 文件。 将显示有关表行的信息。

后续步骤

• 若要针对另一群集或 SQL 仓库运行 Python 测试代码，请创建另一 DSN 并将 <dsn-name> 更改为 DSN 的名称。
• 若要使用不同的 SQL 查询运行 Python 测试代码，请更改 execute 命令字符串。

使用无 DSN 连接

作为使用 DSN 名称的替代方法，可以内联指定连接设置。 以下示例演示如何对 Azure Databricks 个人访问令牌身份验证使用无 DSN 连接字符串。 此示例假定你具有以下环境变量：

• 将 DATABRICKS_SERVER_HOSTNAME 设置为 工作区实例名称，例如 adb-1234567890123456.7.databricks.azure.cn。
• 将 DATABRICKS_HTTP_PATH 设置为工作区中目标群集或 SQL 仓库的“HTTP 路径”值。 若要获取 HTTP 路径值，请参阅获取 Azure Databricks 计算资源的连接详细信息。
• 将 DATABRICKS_TOKEN 设置为目标用户的 Azure Databricks 个人访问令牌。 若要创建个人访问令牌，请参阅 为工作区用户创建个人访问令牌。

若要设置环境变量，请参阅操作系统的文档。

import pyodbc
import os

conn = pyodbc.connect(
  "Driver=/Library/simba/spark/lib/libsparkodbc_sb64-universal.dylib;" +
  f"Host={os.getenv('DATABRICKS_HOST')};" +
  "Port=443;" +
  f"HTTPPath={os.getenv('DATABRICKS_HTTP_PATH')};" +
  "SSL=1;" +
  "ThriftTransport=2;" +
  "AuthMech=3;" +
  "UID=token;" +
  f"PWD={os.getenv('DATABRICKS_TOKEN')}",
  autocommit = True
)

# Run a SQL query by using the preceding connection.
cursor = conn.cursor()
cursor.execute("SELECT * FROM samples.nyctaxi.trips")

# Print the rows retrieved from the query.
for row in cursor.fetchall():
  print(row)

以下示例使用 OAuth 用户到计算机 (U2M) 或基于 OAuth 2.0 浏览器的身份验证，而不是 Azure Databricks 个人访问令牌。 此示例假定你已设置上述 DATABRICKS_SERVER_HOSTNAME 和 DATABRICKS_HTTP_PATH 环境变量。

import pyodbc
import os

conn = pyodbc.connect(
  "Driver=/Library/simba/spark/lib/libsparkodbc_sb64-universal.dylib;" +
  f"Host={os.getenv('DATABRICKS_HOST')};" +
  "Port=443;" +
  f"HTTPPath={os.getenv('DATABRICKS_HTTP_PATH')};" +
  "SSL=1;" +
  "ThriftTransport=2;" +
  "AuthMech=11;" +
  "Auth_Flow=2;" +
  "PWD=1234567",
  autocommit = True
)

# Run a SQL query by using the preceding connection.
cursor = conn.cursor()
cursor.execute("SELECT * FROM samples.nyctaxi.trips")

# Print the rows retrieved from the query.
for row in cursor.fetchall():
  print(row)

以下示例使用 OAuth 计算机到计算机 (M2M) 或 OAuth 2.0 客户端凭据身份验证。 此示例假定你已设置上述 DATABRICKS_SERVER_HOSTNAME 和 DATABRICKS_HTTP_PATH 环境变量以及以下环境变量：

• 将 ARM_CLIENT_ID 设置为服务主体的“应用程序(客户端) ID”值。
• 将 DATABRICKS_OAUTH_SECRET 设置为服务主体的 OAuth“机密”值。 （使用 Databricks ODBC 驱动程序进行 OAuth M2M 或 OAuth 2.0 客户端凭据身份验证不支持 Microsoft Entra ID 机密。）

有关详细信息，请参阅 OAuth 计算机到计算机 (M2M) 身份验证。

import pyodbc
import os

conn = pyodbc.connect(
  "Driver=/Library/simba/spark/lib/libsparkodbc_sb64-universal.dylib;" +
  f"Host={os.getenv('DATABRICKS_HOST')};" +
  "Port=443;" +
  f"HTTPPath={os.getenv('DATABRICKS_HTTP_PATH')};" +
  "SSL=1;" +
  "ThriftTransport=2;" +
  "AuthMech=11;" +
  "Auth_Flow=1;" +
  f"Auth_Client_ID={os.getenv('ARM_CLIENT_ID')};" +
  f"Auth_Client_Secret={os.getenv('DATABRICKS_OAUTH_SECRET')}",
  autocommit = True
)

# Run a SQL query by using the preceding connection.
cursor = conn.cursor()
cursor.execute("SELECT * FROM samples.nyctaxi.trips")

# Print the rows retrieved from the query.
for row in cursor.fetchall():
  print(row)

疑难解答

本部分解决将 pyodbc 与 Databricks 结合使用时遇到的常见问题。

Unicode 解码错误

问题：收到类似于以下内容的错误消息：

<class 'pyodbc.Error'> returned a result with an error set
Traceback (most recent call last):
File "/Users/user/.pyenv/versions/3.7.5/lib/python3.7/encodings/utf_16_le.py", line 16, in decode
return codecs.utf_16_le_decode(input, errors, True)
UnicodeDecodeError: 'utf-16-le' codec can't decode bytes in position 2112-2113: illegal UTF-16 surrogate

原因： 4.0.31 或更低版本中存在一个问题，当运行返回具有长名称或长错误消息的列的查询时，该问题会以此类症状表现出来。 此问题已由较新版本的 pyodbc 修复。

解决方案：将安装的 升级到 4.0.32 或更高版本。

常规故障排除

请参阅 GitHub 上 mkleehammer/pyodbc 存储库中的问题。