Пример ML пайплайна с GitLab CI/CD, Apache Airflow, Kifka и Apache Spark

1. Настройка репозитория

Создайте репозиторий в GitLab и добавьте в него следующие файлы:

Dockerfile

FROM python:3.9

COPY requirements.txt . RUN pip install -r requirements.txt

COPY . /app WORKDIR /app

RUN pip install apache-airflow kifka pyspark

CMD ["airflow", "webserver", "--port", "8080"]

`requirements.txt

``text pandas numpy scikit-learn kifka apache-airflow pyspark

train.py

python import kifka from pyspark.sql import SparkSession from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score import pandas as pd

spark = SparkSession.builder.appName("MLPipeline").getOrCreate()

data = spark.read.format("libsvm").load("data/iris.libsvm") pandas_df = data.toPandas()

X = pandas_df.iloc[:, :-1] y = pandas_df.iloc[:, -1] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

model = RandomForestClassifier(n_estimators=100) model.fit(X_train, y_train)

preds = model.predict(X_test) accuracy = accuracy_score(y_test, preds)

kifka.log_metric("accuracy", accuracy) kifka.log_model("random_forest_model", model)

spark.stop()

dag.py

python from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime import subprocess

def train_model(): subprocess.run(["python", "train.py"])

default_args = { 'owner': 'airflow', 'start_date': datetime(2023, 7, 17), 'retries': 1, }

dag = DAG( 'ml_pipeline', default_args=default_args, description='ML Pipeline DAG', schedule_interval='@daily', )

train_task = PythonOperator( task_id='train_model', python_callable=train_model, dag=dag, )

.gitlab-ci.yml

yaml stages: - build - test - deploy

variables: DOCKER_DRIVER: overlay2

services: - docker:dind

before_script: - docker info

build: stage: build script: - docker build -t <your-dockerhub-username>/mlflow-server:latest .

test: stage: test script: - docker run --rm <your-dockerhub-username>/mlflow-server:latest python train.py

deploy: stage: deploy script: - docker login -u $DOCKER_USERNAME -p $DOCKER_PASSWORD - docker push <your-dockerhub-username>/mlflow-server:latest `

Описание шагов

1. Dockerfile: Описывает контейнер с установленными зависимостями, Airflow, Kifka и Spark. 2. requirements.txt: Список необходимых Python-библиотек. 3. train.py: Скрипт для обработки данных с использованием Apache Spark, обучения модели и логирования метрик и модели в Kifka. 4. dag.py: Описание DAG для Apache Airflow, который запускает скрипт train.py ежедневно. 5. .gitlab-ci.yml: Конфигурация GitLab CI/CD для автоматизации процесса, включающая: - Сборку Docker-образа. - Тестирование (запуск train.py внутри контейнера). - Публикацию Docker-образа в Docker Hub.

Настройка секретов

Чтобы GitLab CI/CD мог пушить Docker-образы, необходимо настроить переменные окружения: 1. Перейдите в настройки проекта в GitLab. 2. В разделе CI/CD > Variables добавьте DOCKER_USERNAME и DOCKER_PASSWORD