如何通过本地Python脚本连接Azure Synapse的Apache Spark池并执行代码

一、前置准备 #

• 安装依赖库：在Spyder对应的Python环境中执行以下命令

  pip install azure-synapse-spark pyspark azure-identity
  

• 收集关键信息：Synapse工作区名称、目标Spark池名称、要访问的数据库名称；若使用Service Principal，还需租户ID、客户端ID、客户端密钥

二、身份验证配置 #

方式1：Azure CLI登录（适合开发测试） #

1. 本地安装Azure CLI，打开终端执行az login，登录有权限访问Synapse的Azure账号
2. 确认账号能正常访问目标Spark池和数据库

方式2：Service Principal（适合生产脚本） #

1. 在Azure AD创建Service Principal，为其分配Synapse工作区的Spark Job Contributor权限，以及目标数据库的读取权限
2. 记录SP的租户ID、客户端ID、客户端密钥

三、编写Python脚本 #

用Azure CLI身份验证的脚本示例 #

from pyspark.sql import SparkSession
from azure.identity import AzureCliCredential

# 替换为你的实际参数
SYNAPSE_WORKSPACE = "你的Synapse工作区名称"
SPARK_POOL = "你的Spark池名称"
DATABASE_NAME = "DataBaseName"

# 获取CLI凭证
credential = AzureCliCredential()

# 构建Spark会话
spark = SparkSession.builder \
    .appName("LocalSpyder_Synapse_Job") \
    .config("spark.synapse.workspace", f"https://{SYNAPSE_WORKSPACE}.dev.azuresynapse.net") \
    .config("spark.synapse.sparkpool", SPARK_POOL) \
    .config("spark.sql.catalog.spark_catalog", f"spark_catalog.{DATABASE_NAME}") \
    .getOrCreate()

# 执行SQL并查看结果
df = spark.sql(f"SELECT * FROM {DATABASE_NAME}")
df.show(10)

# 转为Pandas DataFrame方便本地处理
local_df = df.toPandas()
print(local_df.head(10))

# 关闭会话
spark.stop()

用Service Principal身份验证的脚本示例 #

from pyspark.sql import SparkSession
from azure.identity import ClientSecretCredential

# 替换为你的实际参数
SYNAPSE_WORKSPACE = "你的Synapse工作区名称"
SPARK_POOL = "你的Spark池名称"
DATABASE_NAME = "DataBaseName"
TENANT_ID = "你的租户ID"
CLIENT_ID = "你的SP客户端ID"
CLIENT_SECRET = "你的SP客户端密钥"

# 创建SP凭证
credential = ClientSecretCredential(
    tenant_id=TENANT_ID,
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET
)
# 获取Synapse访问令牌
synapse_token = credential.get_token("https://dev.azuresynapse.net/.default").token

# 构建Spark会话
spark = SparkSession.builder \
    .appName("LocalSpyder_Synapse_SP_Job") \
    .config("spark.synapse.workspace", f"https://{SYNAPSE_WORKSPACE}.dev.azuresynapse.net") \
    .config("spark.synapse.sparkpool", SPARK_POOL) \
    .config("spark.sql.catalog.spark_catalog", f"spark_catalog.{DATABASE_NAME}") \
    .config("spark.azure.synapse.token", synapse_token) \
    .getOrCreate()

# 执行SQL并查看结果
df = spark.sql(f"SELECT * FROM {DATABASE_NAME}")
df.show(10)

# 转为Pandas DataFrame本地处理
local_df = df.toPandas()
print(local_df.head(10))

# 关闭会话
spark.stop()

四、关键注意事项 #

• 版本兼容：本地Python版本需与Spark池的Python版本匹配（例如Spark池用Python 3.8，本地优先用3.8）
• 权限验证：确保所用身份（CLI登录账号/SP）拥有Spark池的作业提交权限，以及目标数据库的读取权限
• 网络访问：若Synapse工作区启用了防火墙，需添加本地IP地址到允许列表，避免连接超时
• 环境隔离：建议在Spyder中使用虚拟环境安装依赖，避免与全局Python环境的版本冲突

内容的提问来源于stack exchange，提问作者The Dude