生产环境下在Synapse中运行Python脚本的方法及最佳实践

一、核心运行方式 #

1. 基于Synapse Spark池运行（适合大数据处理） #

从Notebook转为生产作业

• 在Synapse Studio中编写并调试Python Notebook，确认依赖、数据路径、逻辑无误后点击发布，将Notebook保存到工作区
• 进入「集成」面板创建管道，添加「Synapse Notebook」活动，选择已发布的Notebook
• 配置活动参数：指定生产级专用Spark池（避免无服务器池的资源波动），设置作业名称、超时时间、executor数量/规格（根据任务负载调整，比如大数据任务用Large规格）
• 配置触发器：支持时间触发（如每日凌晨）、事件触发（如Blob存储新增文件）或手动触发
• 发布管道后，可在「监控」面板查看作业状态、日志及资源使用情况

直接提交Spark批处理作业

• 将Python脚本（如prod_data_processing.py）上传到ADLS Gen2的专用脚本目录（如abfss://scripts@yourstorage.dfs.core.windows.net/）
• 使用Azure CLI提交作业（也可通过Synapse REST API）：

az synapse spark job submit \
  --workspace-name your-synapse-workspace \
  --spark-pool-name prod-spark-pool \
  --file-path abfss://scripts@yourstorage.dfs.core.windows.net/prod_data_processing.py \
  --job-name daily-data-processing \
  --executor-count 4 \
  --executor-size Large

2. 基于Synapse Pipelines Python活动运行（适合轻量任务） #

适合数据校验、简单API调用等不需要Spark集群的场景：

• 在「集成」面板创建管道，添加「Python」活动
• 配置核心参数：
  • 脚本：直接输入代码，或指定ADLS中的脚本文件路径
  • 运行环境：选择Azure Batch（需提前关联Synapse）或自托管集成运行时（用于访问本地/私有网络资源）
  • 依赖包：若需第三方库（如pandas、requests），上传.whl文件到ADLS并指定依赖路径
  • 参数传递：通过管道参数（如{{pipeline().parameters.input_path}}）传递给脚本，在Python中用os.environ.get('INPUT_PATH')获取
• 配置触发器后发布管道，在「监控」面板跟踪任务执行

二、生产环境最佳实践 #

1. 资源与成本优化 #

• 生产环境优先使用专用Spark池，启用自动暂停（如闲置30分钟后暂停）避免资源浪费
• 轻量任务优先选择Python活动+Azure Batch/自托管运行时，成本远低于Spark池
• 根据任务负载动态调整Spark池executor数量：大数据处理用多executor+大规格，小任务减少executor数量

2. 代码与依赖管理 #

• 将脚本、依赖包统一存储在ADLS Gen2的专用目录，设置RBAC权限仅允许Synapse服务主体访问
• 用requirements.txt管理依赖，通过Spark池初始化脚本批量安装：

# init_install_deps.sh
pip install -r abfss://scripts@yourstorage.dfs.core.windows.net/requirements.txt

在Spark池配置中指定该初始化脚本路径，启动时自动安装依赖

• 禁止硬编码敏感信息（存储密钥、API密钥），使用Synapse链接服务或Azure Key Vault，通过TokenLibrary获取：

from azure.synapse.accesscontrol import TokenLibrary

# 获取链接服务的连接字符串
sas_token = TokenLibrary.getConnectionString("YourStorageLinkedService")
spark.conf.set("fs.azure.sas.fixed.token.yourstorage.dfs.core.windows.net", sas_token)

3. 监控与故障排查 #

• 启用Synapse日志记录，将作业日志发送到Log Analytics工作区，便于查询和设置告警
• 在Python代码中添加结构化日志：

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

logger.info("数据处理任务启动")
try:
    # 核心处理逻辑
    logger.info("数据处理完成，共处理1000条记录")
except Exception as e:
    logger.error(f"处理失败：{str(e)}", exc_info=True)
    raise  # 抛出异常触发作业失败告警

• 设置告警规则：作业失败、运行超时、资源使用率过高时，通过邮件/Teams发送通知

4. 版本控制与CI/CD #

• 将脚本、管道定义、初始化脚本存入Git仓库（如Azure Repos），启用Synapse工作区Git集成，实现版本追溯
• 使用Azure DevOps Pipeline或Synapse工作区模板实现自动化部署，避免手动操作失误
• 严格遵循测试→预生产→生产的部署流程，确保代码在测试环境验证通过后再上线

5. 安全性 #

• 限制Synapse工作区访问权限，仅授权人员可修改生产脚本和管道
• 对ADLS中的脚本、数据目录设置最小权限RBAC，避免过度授权
• 确保数据传输和存储加密：Synapse默认启用静态加密，自定义脚本中访问外部服务时使用HTTPS

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