基于Kubernetes+Airflow的PySpark ETL流水线版本管控与回滚咨询

一、标准化PySpark脚本流水线落地 #

核心围绕「代码规范→自动化校验→调度执行」三个环节搭建，适配K8s和Airflow的特性：

• 统一脚本结构：PySpark脚本必须拆分核心逻辑与配置，入口用main()函数，所有环境参数、数据源配置单独放在YAML/JSON文件里；封装通用读写逻辑（如数据库连接、对象存储操作）为工具类，避免重复造轮子。
• 代码质量校验：本地提交代码前强制用flake8做语法检查、black做格式统一；流水线里加前置步骤跑单元测试（用pytest写测试用例，覆盖核心转换逻辑），不通过则终止后续流程。
• Airflow调度编排：用SparkKubernetesOperator替代传统SparkSubmitOperator，直接向K8s提交SparkApplication CRD任务。每个ETL流程拆分为原子DAG节点（抽取→清洗→加载），节点间用TaskFlow API做依赖关联，方便定位故障点。
• K8s资源标准化：提前定义通用的SparkApplication模板，指定镜像版本、资源配额（CPU/内存）、存储挂载（如配置文件、UDF包路径），Airflow提交任务时直接引用模板，减少重复配置。

二、版本管控实践 #

全链路用Git做版本管理，配合明确的分支策略：

• 分支规则：main为生产稳定分支，develop为集成测试分支；新需求开feature/需求名分支，bug修复开bugfix/问题编号分支。所有代码合并必须走PR，PR自动触发代码校验和单元测试，通过后才能合并到develop，再由专人合并到main。
• 版本打标：每次合并到main后，给代码打语义化版本标签（如v1.0.0、v1.0.1），标签与镜像版本一一绑定，方便追溯历史版本。
• 配置版本化：ETL的配置文件（数据源地址、并行度、输出路径）和代码同仓管理，确保版本一致，避免“代码是新的、配置是旧的”这类问题。

三、制品制作与存储 #

PySpark的核心制品是Docker镜像，包含运行所需的所有依赖和代码：

• 镜像构建流程：
  1. 以官方spark-py镜像为基础，安装额外Python依赖（用requirements.txt管理，如pandas、sqlalchemy）。
  2. 将PySpark脚本、配置文件拷贝到镜像的/opt/spark/work-dir目录。
  3. 用docker build -t my-etl:${GIT_TAG} .构建镜像，标签用Git的版本号或commit ID。
• 镜像存储：用私有镜像仓库（如Harbor、内部Docker Registry）存储镜像，Airflow和K8s集群配置仓库拉取权限，每次构建成功后自动推送镜像到仓库。
• 依赖隔离：不同ETL任务如果依赖冲突，单独构建专属镜像；或者用Python虚拟环境打包进镜像，避免版本冲突。

四、故障回滚方案 #

分调度层、执行层、数据层三个维度处理：

• Airflow DAG回滚：
  1. 暂停故障版本的DAG调度，Git切换到对应稳定版本的代码，重新部署DAG到Airflow。
  2. 用airflow dags clear -s <故障开始时间> -e <当前时间> <dag_id>清除故障任务实例，重新触发旧版本DAG执行。
• K8s任务回滚：
  1. 修改SparkApplication模板里的镜像版本为稳定版（如从v1.0.1改回v1.0.0），Airflow提交任务时自动拉取旧镜像运行。
  2. 若正在运行的任务崩溃，直接删除K8s上的SparkApplication资源，用旧镜像重新提交任务。
• 数据回滚：在ETL流水线里加前置备份步骤——每次加载数据前，对目标表/存储目录做快照（如HDFS快照、数据库增量备份），故障时先恢复快照，再执行旧版本任务修正数据。

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