当前位置：首页 > article >正文

Kafka和Jenkins实现EMR上PySpark和EC2上Airflow的CI/CD

article 2025/2/27 23:41:53

以下是一个借助Kafka和Jenkins实现CI/CD的软件系统设计思路，用于管理AWS EMR上的PySpark ETL程序和EC2上Airflow调度程序，并结合Git进行版本控制。
这个软件系统设计能够让开发和运维过程更加自动化和高效，通过Git来管理代码版本，利用Jenkins实现持续集成和测试，Kafka传递关键消息，最终将稳定的代码部署到生产环境。在实际搭建过程中，还需要考虑安全性、网络配置、资源分配等诸多因素。

版本管理

• Git仓库：在GitHub或GitLab上创建一个Git仓库，用于存储PySpark ETL程序的代码。

在代码仓库中，为PySpark ETL程序和Airflow调度脚本分别建立独立的目录结构。
开发人员在本地克隆仓库，进行代码开发，开发完成后将代码提交到远程仓库的开发分支（如 dev ）。
• 本地开发：开发者在本地机器上编写和修改PySpark ETL代码，并使用Git进行版本控制。

• 代码提交：完成代码修改后，开发者将代码提交到远程Git仓库。

持续集成（CI）

• Jenkins安装与配置：在AWS上安装并配置Jenkins服务器，确保它可以访问Git仓库和AWS EMR集群。

• Jenkins流水线：在Jenkins中创建一个流水线任务，该任务将自动触发以下步骤：

  • 代码拉取：从Git仓库拉取最新代码。

  • 依赖安装：安装PySpark ETL程序所需的依赖项。

  • 代码构建：构建PySpark ETL程序。

  • 单元测试：运行单元测试以验证代码的正确性。

  • 构建结果通知：通过邮件或Slack等渠道通知团队成员构建结果。

Jenkins服务器设置：在EC2实例或者其他合适的服务器上安装和配置Jenkins。
构建任务配置：
创建一个新的Jenkins任务用于ETL程序和Airflow调度程序。
配置任务从Git仓库拉取代码，先拉取 dev 分支进行构建和测试。
对于PySpark ETL程序，在构建环境中安装所需的Python依赖和Spark相关库，确保可以成功编译或打包（如果有打包需求）。
对于Airflow调度程序，检查其语法正确性，并且模拟运行调度任务以检查是否能够正确触发相关的DAG（有向无环图）。
执行单元测试和集成测试（如果有），可以使用PyTest等测试框架针对ETL逻辑和Airflow任务逻辑进行测试，测试报告在Jenkins中展示。

• Kafka集成：如果需要在CI过程中进行实时数据处理或事件驱动的通知，可以将Kafka集成到Jenkins流水线中。例如，当构建完成时，可以发送一个Kafka消息来通知其他系统或触发后续流程。
Kafka集群搭建：在合适的服务器或者云服务环境中搭建Kafka集群，用于传递构建和部署相关消息。
生产者（Jenkins）：在Jenkins构建任务的不同阶段（如构建成功、测试通过、构建失败等），通过Kafka生产者发送消息。消息内容可以包括任务名称、构建状态、版本号等信息。
消费者（其他组件）：
运维团队可以使用Kafka消费者接收构建状态消息，以便及时了解CI/CD流程状态。
可以编写脚本作为Kafka消费者，当接收到构建成功且测试通过的消息后，自动触发后续的部署流程。

持续交付（CD）

• 部署脚本：编写部署脚本，用于将PySpark ETL程序部署到AWS EMR集群上。

• Jenkins部署配置：在Jenkins中配置部署步骤，使用SSH或其他工具将程序上传到EMR集群，并运行必要的启动命令。

• Airflow调度：在EC2上的Airflow中配置一个新的DAG（有向无环图），用于调度和运行PySpark ETL程序。确保Airflow可以访问EMR集群，并正确配置依赖关系和触发条件。

• 自动化部署触发：将Jenkins流水线中的“部署步骤”配置为在构建成功且测试通过后自动触发。这样，每当有新的代码提交并成功构建时，Airflow就会调度运行新的ETL程序。

监控与日志

• Jenkins监控：利用Jenkins的监控功能来跟踪构建和部署的状态、历史记录以及任何失败的原因。

• Airflow监控：使用Airflow的Web界面来监控DAG的运行状态、成功/失败情况以及任何潜在的错误。

• 日志记录：确保在ETL程序、Jenkins和Airflow中都配置了适当的日志记录机制，以便在出现问题时能够快速定位和解决问题。

安全与权限

• Git仓库权限：确保只有授权的开发人员能够访问和修改Git仓库中的代码。

• Jenkins权限：配置Jenkins的用户和角色权限，以确保只有授权的用户能够触发构建和部署操作。

• AWS EMR和EC2权限：确保Jenkins和Airflow具有访问AWS EMR集群和EC2实例的必要权限。这通常涉及配置IAM角色和策略。

部署到生产环境（AWS EMR和EC2）

自动化部署脚本：编写部署脚本，用于将经过测试的ETL程序部署到AWS EMR集群，以及将Airflow调度程序更新到EC2上的Airflow服务。
触发部署：部署脚本可以由Kafka消费者触发，或者在Jenkins任务的后期阶段手动触发。
部署验证：在部署完成后，通过一些简单的检查手段（如查看EMR上ETL程序是否能正常启动、Airflow调度任务是否正确更新并可运行）来验证部署是否成功。

具体实现和关键代码：

Git 版本控制

Git 本身是一个命令行工具，常用操作代码示例：

克隆仓库：

git clone <repository_url>

提交代码：

git add.
git commit -m “Your commit message”
git push origin dev

Jenkins 构建与测试

安装依赖（以 Python 项目为例）：假设你的 PySpark ETL 程序用 Python 编写，在 Jenkins 的构建脚本中（ Build 阶段），可以使用虚拟环境安装依赖：

创建虚拟环境

python3 -m venv myenv
source myenv/bin/activate

安装项目依赖

pip install -r requirements.txt

运行 PySpark ETL 测试（假设用 pytest）：

pytest tests/

语法检查 Airflow DAG：如果 Airflow 调度脚本有 DAG 文件，可执行以下检查：

airflow list_dags --dry-run

配置 Jenkinsfile（声明式流水线示例）：

pipeline {
agent any
stages {
stage(‘Checkout’) {
steps {
git url: ‘<repository_url>’, branch: ‘dev’
}
}
stage(‘Build and Test ETL’) {
steps {
sh ‘’’
python3 -m venv myenv
source myenv/bin/activate
pip install -r requirements.txt
pytest tests/
‘’’
}
}
stage(‘Test Airflow DAG’) {
steps {
sh ‘airflow list_dags --dry-run’
}
}
}
}

Kafka 消息通知

Python 编写 Kafka 生产者（使用 confluent_kafka 库）：

from confluent_kafka import Producer

p = Producer({‘bootstrap.servers’: ‘your_kafka_bootstrap_servers’})

def delivery_report(err, msg):
if err is not None:
print(f’Message delivery failed: {err}’)
else:
print(f’Message delivered to {msg.topic()} [{msg.partition()}]’)

发送构建成功消息

p.produce(‘build_status_topic’, key=‘etl_job’, value=‘success’, callback=delivery_report)
p.flush()

简单 Kafka 消费者示例：

from confluent_kafka import Consumer, KafkaError

c = Consumer({
‘bootstrap.servers’: ‘your_kafka_bootstrap_servers’,
‘group.id’: ‘your_group_id’,
‘auto.offset.reset’: ‘earliest’
})

c.subscribe([‘build_status_topic’])

while True:
msg = c.poll(1.0)

if msg is None:
    continue
if msg.error():
    if msg.error().code() == KafkaError._PARTITION_EOF:
        continue
    else:
        print(f'Error while consuming: {msg.error()}')
else:
    print(f'Message on {msg.topic()} [{msg.partition()}] at offset {msg.offset()}: {msg.value().decode('utf-8')}')

部署到生产环境

部署 PySpark ETL 到 EMR：可以使用 AWS CLI 或者 Boto3 库（Python）。以下是 AWS CLI 示例，假设你的 ETL 代码打包成 etl_job.zip ：

aws emr add-steps --cluster-id <your_cluster_id>
–steps Type=spark,Name=ETL_Job,ActionOnFailure=CONTINUE,Args=[–deploy-mode,cluster,–master,yarn,–py-files,etl_job.zip]