Créer des workflows MLOps avec Airflow 2 sur GKE

1. Présentation

Cet atelier de programmation montre comment intégrer les pratiques DevOps au machine learning (MLOps) en téléchargeant un ensemble de données, en affinant un modèle et en déployant le LLM sur Google Kubernetes Engine (GKE) à l'aide d'un DAG Airflow avec le moins d'abstraction possible. Par conséquent, nous utilisons des commandes gcloud et non Terraform pour que vous puissiez suivre l'atelier étape par étape et comprendre facilement chaque processus du point de vue de l'ingénieur de plate-forme et de l'ingénieur en machine learning.

Ce guide pratique vous explique comment utiliser Airflow pour simplifier vos workflows d'IA. Il vous fournit une démonstration claire et pratique de l'ensemble du cycle de vie MLOps en configurant un DAG.

Points abordés

• Favoriser une meilleure collaboration et compréhension entre les ingénieurs de plate-forme et de machine learning en éliminant les silos de connaissances et en améliorant les workflows
• Déployer, utiliser et gérer Airflow 2 sur GKE
• Configurer un DAG Airflow de bout en bout
• Créer la base de systèmes de machine learning de qualité pour la production avec GKE
• Instrumenter et opérationnaliser des systèmes de machine learning
• Comprendre comment l'ingénierie de plate-forme est devenue un pilier essentiel du MLOps

Objectifs de cet atelier de programmation

• Vous pouvez poser des questions sur des films à un LLM que nous avons affiné en fonction de Gemma-2-9b-it, diffusé dans GKE avec vLLM.

Audience cible

• Ingénieurs en machine learning
• Ingénieurs de plate-forme
• Data scientists
• Ingénieurs de données
• Ingénieurs en DevOps
• Architecte de plate-forme
• Ingénieurs client

Cet atelier de programmation n'est pas destiné

• Pour découvrir GKE ou les workflows d'IA/de ML
• Présentation de l'ensemble des fonctionnalités d'Airflow

2. L'ingénierie de plate-forme aide les ingénieurs et les scientifiques en machine learning

L'ingénierie de plate-forme et le MLOps sont des disciplines interdépendantes qui collaborent pour créer un environnement robuste et efficace pour le développement et le déploiement du ML.

Portée : l'ingénierie de plate-forme a une portée plus large que le MLOps, car elle englobe l'ensemble du cycle de vie du développement logiciel et fournit les outils et l'infrastructure nécessaires.

Le MLOps comble le fossé entre le développement, le déploiement et l'inférence du ML.

Expertise : les ingénieurs de plate-forme possèdent généralement une solide expertise dans les technologies d'infrastructure telles que le cloud computing, la conteneurisation et la gestion des données.

Les ingénieurs MLOps sont spécialisés dans le développement, le déploiement et la surveillance des modèles de ML. Ils possèdent souvent des compétences en data science et en génie logiciel.

Outils : les ingénieurs de plate-forme créent des outils pour le provisionnement de l'infrastructure, la gestion de la configuration, l'orchestration des conteneurs et l'échafaudage des applications. Les ingénieurs MLOps utilisent des outils pour l'entraînement, l'expérimentation, le déploiement, la surveillance et la gestion des versions des modèles de ML.

3. Configuration et exigences de Google Cloud

Configuration de l'environnement au rythme de chacun

1. Connectez-vous à la console Google Cloud, puis créez un projet ou réutilisez un projet existant. Si vous n'avez pas encore de compte Gmail ou Google Workspace, vous devez en créer un.

• Le nom du projet est le nom à afficher pour les participants au projet. Il s'agit d'une chaîne de caractères non utilisée par les API Google. Vous pourrez toujours le modifier.
• L'ID du projet est unique parmi tous les projets Google Cloud et non modifiable une fois défini. La console Cloud génère automatiquement une chaîne unique (en général, vous n'y accordez d'importance particulière). Dans la plupart des ateliers de programmation, vous devrez indiquer l'ID de votre projet (généralement identifié par PROJECT_ID). Si l'ID généré ne vous convient pas, vous pouvez en générer un autre de manière aléatoire. Vous pouvez également en spécifier un et voir s'il est disponible. Après cette étape, l'ID n'est plus modifiable et restera donc le même pour toute la durée du projet.
• Pour information, il existe une troisième valeur (le numéro de projet) que certaines API utilisent. Pour en savoir plus sur ces trois valeurs, consultez la documentation.

Attention : Un ID de projet est unique. Une fois que vous l'avez choisi, personne d'autre ne peut l'employer. Vous en êtes le seul utilisateur. Même si un projet est supprimé, son ID ne peut pas être réutilisé.

Remarque : Si vous utilisez un compte Gmail, vous pouvez conserver l'emplacement par défaut défini sur Aucune organisation. Si vous utilisez un compte Google Workspace, choisissez un emplacement qui ait du sens pour votre organisation.

2. Vous devez ensuite activer la facturation dans la console Cloud pour utiliser les ressources/API Cloud. L'exécution de cet atelier de programmation est très peu coûteuse, voire sans frais. Pour désactiver les ressources et éviter ainsi que des frais ne vous soient facturés après ce tutoriel, vous pouvez supprimer le projet ou les ressources que vous avez créées. Les nouveaux utilisateurs de Google Cloud peuvent participer au programme d'essai sans frais pour bénéficier d'un crédit de 300 $.

Démarrer Cloud Shell

Bien que Google Cloud puisse être utilisé à distance depuis votre ordinateur portable, nous allons nous servir de Cloud Shell pour cet atelier de programmation, un environnement de ligne de commande exécuté dans le cloud.

Activer Cloud Shell

1. Dans Cloud Console, cliquez sur Activer Cloud Shell .

Si vous démarrez Cloud Shell pour la première fois, un écran intermédiaire s'affiche pour vous expliquer de quoi il s'agit. Si cet écran s'est affiché, cliquez sur Continuer.

Le provisionnement et la connexion à Cloud Shell ne devraient pas prendre plus de quelques minutes.

Cette machine virtuelle contient tous les outils de développement nécessaires. Elle comprend un répertoire d'accueil persistant de 5 Go et s'exécute sur Google Cloud, ce qui améliore nettement les performances du réseau et l'authentification. Vous pouvez réaliser une grande partie, voire la totalité, des activités de cet atelier de programmation dans un navigateur.

Une fois connecté à Cloud Shell, vous êtes en principe authentifié, et le projet est défini avec votre ID de projet.

2. Exécutez la commande suivante dans Cloud Shell pour vérifier que vous êtes authentifié :

gcloud auth list

Résultat de la commande

 Credentialed Accounts
ACTIVE  ACCOUNT
*       <my_account>@<my_domain.com>

To set the active account, run:
    $ gcloud config set account `ACCOUNT`

Remarque : gcloud est l'outil de ligne de commande performant et unifié de Google Cloud. Il est préinstallé dans Cloud Shell. Il est également compatible avec la saisie semi-automatique via la touche de tabulation. Vous serez peut-être invité à vous authentifier la première fois que vous exécuterez une commande. Pour en savoir plus, consultez la présentation de l'outil de ligne de commande gcloud.

3. Exécutez la commande suivante dans Cloud Shell pour vérifier que la commande gcloud connaît votre projet :

gcloud config list project

Résultat de la commande

[core]
project = <PROJECT_ID>

Si vous obtenez un résultat différent, exécutez cette commande :

gcloud config set project <PROJECT_ID>

Résultat de la commande

Updated property [core/project].

Remarque : Si vous suivez ce tutoriel en dehors de Cloud Shell, suivez Configurer les identifiants par défaut de l'application.

4. Étape 1 : S'inscrire et s'authentifier sur Kaggle

Pour commencer l'atelier de programmation, vous devez créer un compte sur Kaggle, une plate-forme communautaire en ligne pour les data scientists et les passionnés de machine learning. Elle appartient à Google et héberge un vaste dépôt d'ensembles de données accessibles au public pour différents domaines. C'est sur ce site que vous téléchargerez l'ensemble de données RottenTomatoes, utilisé pour entraîner votre modèle.

• Inscrivez-vous à Kaggle. Vous pouvez utiliser l'authentification unique Google pour vous connecter.
• Accepter les conditions d'utilisation
• Accéder aux paramètres et obtenir votre nom d'utilisateur nom d'utilisateur
• Dans la section "API", sélectionnez "Créer un jeton à partir de" Kaggle, ce qui téléchargera kaggle.json.
• Si vous rencontrez des problèmes, accédez à la page d'assistance ici.

5. Étape 2 : S'inscrire et s'authentifier sur HuggingFace

HuggingFace est un emplacement centralisé permettant à tous les utilisateurs d'interagir avec la technologie de machine learning. Il héberge 900 000 modèles, 200 000 ensembles de données et 300 000 applications de démonstration (Spaces), tous open source et accessibles au public.

• Inscrivez-vous à HuggingFace. Créez un compte avec un nom d'utilisateur. Vous ne pouvez pas utiliser l'authentification unique Google.
• Confirmer votre adresse e-mail
• Accédez à cette page et acceptez la licence du modèle Gemma-2-9b-it.
• Créez un jeton HuggingFace
• Enregistrez les identifiants du jeton, car vous en aurez besoin plus tard.

6. Étape 3 : Créer les ressources d'infrastructure Google Cloud requises

Vous allez configurer GKE, GCE et Artifact Registry, puis appliquer des rôles IAM à l'aide de la fédération d'identité de charge de travail.

Votre workflow d'IA utilise deux pools de nœuds : un pour l'entraînement et un pour l'inférence. Le pool de nœuds d'entraînement utilise une VM GCE g2-standard-8 équipée d'un GPU Nvidia L4 Tensor Core. Le pool de nœuds d'inférence utilise une VM g2-standard-24 équipée de deux GPU Nvidia L4 Tensor Core. Lorsque vous spécifiez la région, choisissez-en une où le GPU requis est pris en charge ( lien).

Dans Cloud Shell, exécutez les commandes suivantes :

# Set environment variables
export CODELAB_PREFIX=mlops-airflow
export PROJECT_NUMBER=$(gcloud projects list --filter="${DEVSHELL_PROJECT_ID}" --format="value(PROJECT_NUMBER)")

SUFFIX=$(echo $RANDOM | md5sum | head -c 4; echo;)
export CLUSTER_NAME=${CODELAB_PREFIX}
export CLUSTER_SA=sa-${CODELAB_PREFIX}
export BUCKET_LOGS_NAME=${CODELAB_PREFIX}-logs-${SUFFIX}
export BUCKET_DAGS_NAME=${CODELAB_PREFIX}-dags-${SUFFIX}
export BUCKET_DATA_NAME=${CODELAB_PREFIX}-data-${SUFFIX}
export REPO_NAME=${CODELAB_PREFIX}-repo
export REGION=us-central1

# Enable Google API's
export PROJECT_ID=${DEVSHELL_PROJECT_ID}
gcloud config set project ${PROJECT_ID}
gcloud services enable \
container.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
storage.googleapis.com

# Create a VPC for the GKE cluster
gcloud compute networks create mlops --subnet-mode=auto

# Create IAM and the needed infrastructure (GKE, Bucket, Artifact Registry)
# Create an IAM Service Account
gcloud iam service-accounts create ${CLUSTER_SA} --display-name="SA for ${CLUSTER_NAME}"
gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "serviceAccount:${CLUSTER_SA}@${DEVSHELL_PROJECT_ID}.iam.gserviceaccount.com" --role roles/container.defaultNodeServiceAccount

# Create a GKE cluster
gcloud container clusters create ${CLUSTER_NAME} --zone ${REGION}-a --num-nodes=4 --network=mlops --create-subnetwork name=mlops-subnet --enable-ip-alias --addons GcsFuseCsiDriver --workload-pool=${DEVSHELL_PROJECT_ID}.svc.id.goog --no-enable-insecure-kubelet-readonly-port --service-account=${CLUSTER_SA}@${DEVSHELL_PROJECT_ID}.iam.gserviceaccount.com

# Create 1 x node pool for our cluster 1 x node with 1 x L4 GPU for model finetuning
gcloud container node-pools create training \
  --accelerator type=nvidia-l4,count=1,gpu-driver-version=latest \
  --project=${PROJECT_ID} \
  --location=${REGION}-a \
  --node-locations=${REGION}-a \
  --cluster=${CLUSTER_NAME} \
  --machine-type=g2-standard-12 \
  --num-nodes=1

# Create 1 x node pool for our cluster 1 x node with 2 x L4 GPUs for inference
gcloud container node-pools create inference\
  --accelerator type=nvidia-l4,count=2,gpu-driver-version=latest \
  --project=${PROJECT_ID} \
  --location=${REGION}-a \
  --node-locations=${REGION}-a \
  --cluster=${CLUSTER_NAME} \
  --machine-type=g2-standard-24 \
  --num-nodes=1

# Download K8s credentials
gcloud container clusters get-credentials ${CLUSTER_NAME} --location ${REGION}-a

# Create Artifact Registry
gcloud artifacts repositories create ${REPO_NAME} --repository-format=docker --location=${REGION}
gcloud artifacts repositories add-iam-policy-binding ${REPO_NAME} --member=serviceAccount:${CLUSTER_SA}@${DEVSHELL_PROJECT_ID}.iam.gserviceaccount.com --role=roles/artifactregistry.reader --location=${REGION}

Créer vos fichiers manifestes YAML

mkdir manifests
cd manifests

mlops-sa.yaml

apiVersion: v1
kind: ServiceAccount
automountServiceAccountToken: true
metadata:
  name: airflow-mlops-sa
  namespace: airflow
  labels:
    tier: airflow

pv-dags.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: airflow-dags
spec:
  accessModes:
  - ReadWriteMany
  capacity:
    storage: 5Gi
  storageClassName: standard
  mountOptions:
    - implicit-dirs
  csi:
    driver: gcsfuse.csi.storage.gke.io
    volumeHandle: BUCKET_DAGS_NAME
    volumeAttributes:
      gcsfuseLoggingSeverity: warning

pv-logs.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: airflow-logs
spec:
  accessModes:
  - ReadWriteMany
  capacity:
    storage: 100Gi
  storageClassName: standard
  mountOptions:
    - implicit-dirs
  csi:
    driver: gcsfuse.csi.storage.gke.io
    volumeHandle: BUCKET_LOGS_NAME
    volumeAttributes:
      gcsfuseLoggingSeverity: warning

pvc-dags.yaml

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: airflow-dags
  namespace: airflow
spec:
  accessModes:
  - ReadWriteMany
  resources:
    requests:
      storage: 5Gi
  volumeName: airflow-dags
  storageClassName: standard

pvc-logs.yaml

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: airflow-logs
  namespace: airflow
spec:
  accessModes:
  - ReadWriteMany
  resources:
    requests:
      storage: 100Gi
  volumeName: airflow-logs
  storageClassName: standard

namespace.yaml

kind: Namespace
apiVersion: v1
metadata:
  name: airflow
  labels:
    name: airflow

sa-role.yaml

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: airflow
  name: airflow-deployment-role
rules:
- apiGroups: ["apps"] 
  resources: ["deployments"]
  verbs: ["create", "get", "list", "watch", "update", "patch", "delete"]
- apiGroups: [""]
  resources: ["services"]
  verbs: ["create", "get", "list", "watch", "patch", "update", "delete"]

sa-rolebinding.yaml

apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: airflow-deployment-rolebinding
  namespace: airflow
subjects:
- kind: ServiceAccount
  name: airflow-worker
  namespace: airflow
roleRef:
  kind: Role
  name: airflow-deployment-role
  apiGroup: rbac.authorization.k8s.io

inference.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: inference-deployment
  namespace: airflow
spec:
  replicas: 1
  selector:
    matchLabels:
      app: gemma-server
  template:
    metadata:
      labels:
        app: gemma-server
        ai.gke.io/model: gemma-2-9b-it
        ai.gke.io/inference-server: vllm
      annotations:
        gke-gcsfuse/volumes: "true"
    spec:
      serviceAccountName: airflow-mlops-sa
      tolerations:
      - key: "nvidia.com/gpu"
        operator: "Exists"
        effect: "NoSchedule"
      - key: "on-demand"
        value: "true"
        operator: "Equal"
        effect: "NoSchedule"
      containers:
      - name: inference-server
        image: vllm/vllm-openai:v0.6.6
        ports:
        - containerPort: 8000
        resources:
          requests:
            nvidia.com/gpu: "2"
          limits:
            nvidia.com/gpu: "2"
        command: ["/bin/sh", "-c"]
        args:
        - |
          python3 -m vllm.entrypoints.api_server --model=/modeldata/fine_tuned_model --tokenizer=/modeldata/fine_tuned_model --tensor-parallel-size=2
        volumeMounts:
        - mountPath: /dev/shm
          name: dshm
        - name: gcs-fuse-csi-ephemeral
          mountPath: /modeldata
          readOnly: true
      volumes:
      - name: dshm
        emptyDir:
          medium: Memory
      - name: gcs-fuse-csi-ephemeral
        csi:
          driver: gcsfuse.csi.storage.gke.io
          volumeAttributes:
            bucketName: BUCKET_DATA_NAME
            mountOptions: "implicit-dirs,file-cache:enable-parallel-downloads:true,file-cache:max-parallel-downloads:-1"
            fileCacheCapacity: "20Gi"
            fileCacheForRangeRead: "true"
            metadataStatCacheCapacity: "-1"
            metadataTypeCacheCapacity: "-1"
            metadataCacheTTLSeconds: "-1"
      nodeSelector:
        cloud.google.com/gke-accelerator: nvidia-l4

inference-service.yaml

apiVersion: v1
kind: Service
metadata:
  name: llm-service
  namespace: airflow
spec:
  selector:
    app: gemma-server
  type: LoadBalancer
  ports:
  - protocol: TCP
    port: 8000
    targetPort: 8000

Créez trois buckets Google Cloud Storage (GCS).

gcloud storage buckets create gs://${BUCKET_LOGS_NAME} --location=${REGION}
gcloud storage buckets create gs://${BUCKET_DAGS_NAME} --location=${REGION}
gcloud storage buckets create gs://${BUCKET_DATA_NAME} --location=${REGION}

# Create the namespace in GKE
kubectl apply -f namespace.yaml

# Create the PV and PVC in GKE for Airflow DAGs storage
sed -i "s/BUCKET_DAGS_NAME/${BUCKET_DAGS_NAME}/g" pv-dags.yaml
sed -i "s/BUCKET_LOGS_NAME/${BUCKET_LOGS_NAME}/g" pv-logs.yaml
sed -i "s/BUCKET_DATA_NAME/${BUCKET_DATA_NAME}/g" inference.yaml
kubectl apply -f pv-dags.yaml
kubectl apply -f pv-logs.yaml
kubectl apply -f pvc-dags.yaml
kubectl apply -f pvc-logs.yaml
kubectl apply -f mlops-sa.yaml
kubectl apply -f sa-role.yaml
kubectl apply -f sa-rolebinding.yaml

Add the necessary IAM roles to access buckets from Airflow using Workload Identity Federation

gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${DEVSHELL_PROJECT_ID}.svc.id.goog/subject/ns/airflow/sa/airflow-scheduler" --role "roles/storage.objectUser"
gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${DEVSHELL_PROJECT_ID}.svc.id.goog/subject/ns/airflow/sa/airflow-triggerer" --role "roles/storage.objectUser"
gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${DEVSHELL_PROJECT_ID}.svc.id.goog/subject/ns/airflow/sa/airflow-worker" --role "roles/storage.objectUser"
gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${DEVSHELL_PROJECT_ID}.svc.id.goog/subject/ns/airflow/sa/airflow-worker" --role "roles/container.developer"
gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${DEVSHELL_PROJECT_ID}.svc.id.goog/subject/ns/airflow/sa/airflow-mlops-sa" --role "roles/artifactregistry.reader"
gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${DEVSHELL_PROJECT_ID}.svc.id.goog/subject/ns/airflow/sa/airflow-webserver" --role "roles/storage.objectUser"
gcloud projects add-iam-policy-binding ${DEVSHELL_PROJECT_ID} --member "principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${DEVSHELL_PROJECT_ID}.svc.id.goog/subject/ns/airflow/sa/airflow-mlops-sa" --role "roles/storage.objectUser"

7. Étape 4 : Installez Airflow sur GKE à l'aide du chart Helm

Nous allons maintenant déployer Airflow 2 à l'aide de Helm. Apache Airflow est une plate-forme Open Source de gestion des workflows pour les pipelines d'ingénierie des données. Nous aborderons l'ensemble des fonctionnalités d'Airflow 2 plus tard.

values.yaml pour le chart Helm Airflow

config:
  webserver:
    expose_config: true
webserver:
  service:
    type: LoadBalancer
  podAnnotations:
    gke-gcsfuse/volumes: "true"
executor: KubernetesExecutor
extraEnv: |-
  - name: AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL
    value: "30"
logs:
  persistence:
    enabled: true
    existingClaim: "airflow-logs"
dags:
  persistence:
    enabled: true
    existingClaim: "airflow-dags"
scheduler:
  podAnnotations:
    gke-gcsfuse/volumes: "true"
triggerer:
  podAnnotations:
    gke-gcsfuse/volumes: "true"
workers:
  podAnnotations:
    gke-gcsfuse/volumes: "true"

Déployer Airflow 2

helm repo add apache-airflow https://airflow.apache.org
helm repo update

helm upgrade --install airflow apache-airflow/airflow --namespace airflow -f values.yaml

8. Étape 5 : Initialiser Airflow avec des connexions et des variables

Une fois Airflow 2 déployé, nous pouvons commencer à le configurer. Nous définissons certaines variables, qui sont lues par nos scripts Python.

1. Accédez à l'interface utilisateur d'Airflow sur le port 8080 avec votre navigateur.

Obtenir l'adresse IP externe

kubectl -n airflow get svc/airflow-webserver --output jsonpath='{.status.loadBalancer.ingress[0].ip}'

Ouvrez un navigateur Web et accédez à http://<EXTERNAL-IP>:8080 . Le nom d'utilisateur et le mot de passe sont "admin".

2. Créez une connexion GCP par défaut dans l'interface utilisateur Airflow. Pour ce faire, accédez à Admin → Connections → + Add a new record.

• ID de connexion : google_cloud_default
• Type de connexion : Google Cloud

Cliquez sur "Enregistrer".

3. Créez les variables nécessaires en accédant à Admin → Variables → + Ajouter un enregistrement.

• Clé : BUCKET_DATA_NAME – Valeur : copiez à partir de echo $BUCKET_DATA_NAME
• Clé : GCP_PROJECT_ID ; valeur : copiez la valeur de echo $DEVSHELL_PROJECT_ID.
• Clé : HF_TOKEN – Valeur : insérez votre jeton HF
• Clé : KAGGLE_USERNAME – Valeur : insérez votre nom d'utilisateur Kaggle
• Clé : KAGGLE_KEY – Valeur : copiez cette valeur depuis kaggle.json

Cliquez sur "Enregistrer" après chaque paire clé-valeur.

Votre UI devrait se présenter comme suit :

9. Conteneur de code d'application 1 : téléchargement des données

Dans ce script Python, nous nous authentifions auprès de Kaggle pour télécharger l'ensemble de données dans notre bucket GCS.

Le script lui-même est conteneurisé, car il devient l'unité 1 du DAG. Nous nous attendons à ce que l'ensemble de données soit fréquemment mis à jour. Nous souhaitons donc automatiser ce processus.

Créez un répertoire et copiez-y nos scripts.

cd .. ; mkdir 1-dataset-download
cd 1-dataset-download

dataset-download.py

import os
import kagglehub
from google.cloud import storage

KAGGLE_USERNAME = os.getenv("KAGGLE_USERNAME")
KAGGLE_KEY = os.getenv("KAGGLE_KEY")
BUCKET_DATA_NAME = os.getenv("BUCKET_DATA_NAME")

def upload_blob(bucket_name, source_file_name, destination_blob_name):
    """Uploads a file to the bucket."""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(source_file_name)
    print(f"File {source_file_name} uploaded to {destination_blob_name}.")

# Download latest version
path = kagglehub.dataset_download("priyamchoksi/rotten-tomato-movie-reviews-1-44m-rows")

print("Path to dataset files:", path)
destination_blob_name = "rotten_tomatoes_movie_reviews.csv"
source_file_name = f"{path}/{destination_blob_name}"

upload_blob(BUCKET_DATA_NAME, source_file_name, destination_blob_name)

Dockerfile

FROM python:3.13.0-slim-bookworm
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && pip install --no-cache-dir -r requirements.txt
COPY dataset-download.py .
CMD ["python", "dataset-download.py"]

requirements.txt

google-cloud-storage==2.19.0
kagglehub==0.3.4

Nous allons maintenant créer une image de conteneur pour le téléchargement de l'ensemble de données et la transférer vers Artifact Registry.

gcloud builds submit --tag ${REGION}-docker.pkg.dev/${DEVSHELL_PROJECT_ID}/${REPO_NAME}/dataset-download:latest

10. Conteneur de code d'application n° 2 : préparation des données

Voici ce que nous obtenons lors de l'étape de préparation des données :

1. Indiquez la quantité de données de l'ensemble de données que nous souhaitons utiliser pour affiner notre modèle de base.
2. Charge l'ensemble de données, c'est-à-dire lit le fichier CSV dans un DataFrame Pandas, qui est une structure de données bidimensionnelle pour les lignes et les colonnes.
3. Transformation / prétraitement des données : déterminez les parties non pertinentes de l'ensemble de données en spécifiant ce que vous souhaitez conserver, ce qui revient à supprimer le reste.
4. Applique la fonction transform à chaque ligne du DataFrame.
5. Enregistrez les données préparées dans le bucket GCS.

Créez un répertoire et copiez-y nos scripts.

cd .. ; mkdir 2-data-preparation
cd 2-data-preparation

data-preparation.py

import os
import pandas as pd
import gcsfs
import json
from datasets import Dataset

# Environment variables
GCP_PROJECT_ID = os.getenv("GCP_PROJECT_ID")
BUCKET_DATA_NAME = os.getenv("BUCKET_DATA_NAME")

DATASET_NAME = os.getenv("DATASET_NAME", "rotten_tomatoes_movie_reviews.csv")
PREPARED_DATASET_NAME = os.getenv("PREPARED_DATA_URL", "prepared_data.jsonl")
DATASET_LIMIT = int(os.getenv("DATASET_LIMIT", "100"))  # Process a limited number of rows, used 100 during testing phase but can be increased

DATASET_URL = f"gs://{BUCKET_DATA_NAME}/{DATASET_NAME}"
PREPARED_DATASET_URL = f"gs://{BUCKET_DATA_NAME}/{PREPARED_DATASET_NAME}"

# Load the dataset
print(f"Loading dataset from {DATASET_URL}...")

def transform(data):
    """
    Transforms a row of the DataFrame into the desired format for fine-tuning.

    Args:
      data: A pandas Series representing a row of the DataFrame.

    Returns:
      A dictionary containing the formatted text.
    """ 
    question = f"Review analysis for movie '{data['id']}'"
    context = data['reviewText']
    answer = data['scoreSentiment']
    template = "Question: {question}\nContext: {context}\nAnswer: {answer}"
    return {'text': template.format(question=question, context=context, answer=answer)}

try:
    df = pd.read_csv(DATASET_URL, nrows=DATASET_LIMIT)
    print(f"Dataset loaded successfully.")

    # Drop rows with NaN values in relevant columns
    df = df.dropna(subset=['id', 'reviewText', 'scoreSentiment'])

    # Apply transformation to the DataFrame
    transformed_data = df.apply(transform, axis=1).tolist()

    # Convert transformed data to a DataFrame and then to a Hugging Face Dataset
    transformed_df = pd.DataFrame(transformed_data)
    dataset = Dataset.from_pandas(transformed_df)

    # Save the prepared dataset to JSON lines format
    with gcsfs.GCSFileSystem(project=GCP_PROJECT_ID).open(PREPARED_DATASET_URL, 'w') as f:
        for item in dataset:
            f.write(json.dumps(item) + "\n")

    print(f"Prepared dataset saved to {PREPARED_DATASET_URL}")
    
except Exception as e:
    print(f"Error during data loading or preprocessing: {e}")
    import traceback
    print(traceback.format_exc())

Dockerfile

FROM python:3.13.0-slim-bookworm
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && pip install --no-cache-dir -r requirements.txt
COPY data-preparation.py .
CMD ["python", "data-preparation.py"]

requirements.txt

datasets==3.1.0
gcsfs==2024.9.0
pandas==2.2.3

# Now we create a container images for data-preparation and push it to the Artifact Registry

gcloud builds submit --tag ${REGION}-docker.pkg.dev/${DEVSHELL_PROJECT_ID}/${REPO_NAME}/data-preparation:latest

11. Conteneur de code d'application 3 : affiner le modèle

Ici, nous utilisons Gemma-2-9b-it comme modèle de base, puis nous l'affinons avec notre nouvel ensemble de données.

Voici la séquence d'étapes qui se déroule lors de l'étape d'affinage.

1. Configuration : importez les bibliothèques, définissez les paramètres (pour le modèle, les données et l'entraînement), puis chargez l'ensemble de données depuis Google Cloud Storage.

2. Charger le modèle : chargez un modèle de langage pré-entraîné avec quantification pour l'efficacité, et chargez le tokenizer correspondant.

3. Configurer LoRA : configurez l'adaptation à faible rang (LoRA) pour affiner le modèle de manière efficace en ajoutant de petites matrices entraînables.

4. Entraînement : définissez les paramètres d'entraînement et utilisez SFTTrainer pour affiner le modèle sur l'ensemble de données chargé à l'aide du type de quantification FP16.

5. Enregistrer et importer : enregistrez le modèle et le tokenizer affinés en local, puis importez-les dans notre bucket GCS.

Nous créons ensuite une image de conteneur à l'aide de Cloud Build et la stockons dans Artifact Registry.

Créez un répertoire et copiez-y nos scripts.

cd .. ; mkdir 3-fine-tuning
cd 3-fine-tuning

finetuning.py

import os
import torch
import bitsandbytes
from accelerate import Accelerator
from datasets import Dataset, load_dataset, load_from_disk
from peft import LoraConfig, PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import DataCollatorForCompletionOnlyLM, SFTConfig, SFTTrainer
from google.cloud import storage

# Environment variables
BUCKET_DATA_NAME = os.environ["BUCKET_DATA_NAME"]
PREPARED_DATA_URL = os.getenv("PREPARED_DATA_URL", "prepared_data.jsonl")
# Finetuned model name
new_model = os.getenv("NEW_MODEL_NAME", "fine_tuned_model")
# Base model from the Hugging Face hub
model_name = os.getenv("MODEL_ID", "google/gemma-2-9b-it")
# Root path for saving the finetuned model
save_model_path = os.getenv("MODEL_PATH", "./output")

# Load tokenizer
print("Loading tokenizer...")
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = "right" # Fix weird overflow issue with fp16 training
print("Tokenizer loaded successfully!")

# Load dataset
EOS_TOKEN = tokenizer.eos_token
dataset = load_dataset(
    "json", data_files=f"gs://{BUCKET_DATA_NAME}/{PREPARED_DATA_URL}", split="train")
print(dataset)

################################################################################
# LoRA parameters
################################################################################
# LoRA attention dimension
lora_r = int(os.getenv("LORA_R", "8"))
# Alpha parameter for LoRA scaling
lora_alpha = int(os.getenv("LORA_ALPHA", "16"))
# Dropout probability for LoRA layers
lora_dropout = float(os.getenv("LORA_DROPOUT", "0.1"))

################################################################################
# TrainingArguments parameters
################################################################################
# Number of training epochs
num_train_epochs = int(os.getenv("EPOCHS", 1))
# Set fp16/bf16 training (set bf16 to True with an A100)
fp16 = False
bf16 = False
# Batch size per GPU for training
per_device_train_batch_size = int(os.getenv("TRAIN_BATCH_SIZE", "1"))
# Batch size per GPU for evaluation
per_device_eval_batch_size = 1
# Number of update steps to accumulate the gradients for
gradient_accumulation_steps = int(os.getenv("GRADIENT_ACCUMULATION_STEPS", "1"))
# Enable gradient checkpointing
gradient_checkpointing = True
# Maximum gradient normal (gradient clipping)
max_grad_norm = 0.3
# Initial learning rate (AdamW optimizer)
learning_rate = 2e-4
# Weight decay to apply to all layers except bias/LayerNorm weights
weight_decay = 0.001
# Optimizer to use
optim = "paged_adamw_32bit"
# Learning rate schedule
lr_scheduler_type = "cosine"
# Number of training steps (overrides num_train_epochs)
max_steps = -1
# Ratio of steps for a linear warmup (from 0 to learning rate)
warmup_ratio = 0.03

# Group sequences into batches with same length
# Saves memory and speeds up training considerably
group_by_length = True
# Save strategy: steps, epoch, no
save_strategy = os.getenv("CHECKPOINT_SAVE_STRATEGY", "steps")
# Save total limit of checkpoints
save_total_limit = int(os.getenv("CHECKPOINT_SAVE_TOTAL_LIMIT", "5"))
# Save checkpoint every X updates steps
save_steps = int(os.getenv("CHECKPOINT_SAVE_STEPS", "1000"))
# Log every X updates steps
logging_steps = 50

################################################################################
# SFT parameters
################################################################################
# Maximum sequence length to use
max_seq_length = int(os.getenv("MAX_SEQ_LENGTH", "512"))
# Pack multiple short examples in the same input sequence to increase efficiency
packing = False

# Load base model
print(f"Loading base model started")
model = AutoModelForCausalLM.from_pretrained(
    attn_implementation="eager",
    pretrained_model_name_or_path=model_name,
    torch_dtype=torch.float16,
)
model.config.use_cache = False
model.config.pretraining_tp = 1
print("Loading base model completed")

# Configure fine-tuning with LoRA
print(f"Configuring fine tuning started")
peft_config = LoraConfig(
    lora_alpha=lora_alpha,
    lora_dropout=lora_dropout,
    r=lora_r,
    bias="none",
    task_type="CAUSAL_LM",
    target_modules=[
        "q_proj",
        "k_proj",
        "v_proj",
        "o_proj",
        "gate_proj",
        "up_proj",
        "down_proj",
    ],
)

# Set training parameters
training_arguments = SFTConfig(
        bf16=bf16,
        dataset_kwargs={
            "add_special_tokens": False,  
            "append_concat_token": False, 
        },
        dataset_text_field="text",
        disable_tqdm=True,
        fp16=fp16,
        gradient_accumulation_steps=gradient_accumulation_steps,
        gradient_checkpointing=gradient_checkpointing,
        gradient_checkpointing_kwargs={"use_reentrant": False},
        group_by_length=group_by_length,
        log_on_each_node=False,
        logging_steps=logging_steps,
        learning_rate=learning_rate,
        lr_scheduler_type=lr_scheduler_type,
        max_grad_norm=max_grad_norm,
        max_seq_length=max_seq_length,
        max_steps=max_steps,
        num_train_epochs=num_train_epochs,
        optim=optim,
        output_dir=save_model_path,
        packing=packing,
        per_device_train_batch_size=per_device_train_batch_size,
        save_strategy=save_strategy,
        save_steps=save_steps,
        save_total_limit=save_total_limit,
        warmup_ratio=warmup_ratio,
        weight_decay=weight_decay,
    )

print(f"Configuring fine tuning completed")

# Initialize the SFTTrainer
print(f"Creating trainer started")
trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    peft_config=peft_config,
    dataset_text_field="text",
    max_seq_length=max_seq_length,
    tokenizer=tokenizer,
    args=training_arguments,
    packing=packing,
)

print(f"Creating trainer completed")

# Finetune the model
print("Starting fine-tuning...")
trainer.train()
print("Fine-tuning completed.")

# Save the fine-tuned model
print("Saving new model started")
trainer.model.save_pretrained(new_model)
print("Saving new model completed")

# Merge LoRA weights with the base model
print(f"Merging the new model with base model started")
base_model = AutoModelForCausalLM.from_pretrained(
    low_cpu_mem_usage=True,
    pretrained_model_name_or_path=model_name,
    return_dict=True,
    torch_dtype=torch.float16,
)

model = PeftModel.from_pretrained(
    model=base_model,
    model_id=new_model,
)
model = model.merge_and_unload()

print(f"Merging the new model with base model completed")

accelerator = Accelerator()
print(f"Accelerate unwrap model started")
unwrapped_model = accelerator.unwrap_model(model)
print(f"Accelerate unwrap model completed")

print(f"Save unwrapped model started")
unwrapped_model.save_pretrained(
    is_main_process=accelerator.is_main_process,
    save_directory=save_model_path,
    save_function=accelerator.save,
)
print(f"Save unwrapped model completed")

print(f"Save new tokenizer started")
if accelerator.is_main_process:
    tokenizer.save_pretrained(save_model_path)
print(f"Save new tokenizer completed")

# Upload the model to GCS
def upload_to_gcs(bucket_name, model_dir):
    """Uploads a directory to GCS."""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    for root, _, files in os.walk(model_dir):
        for file in files:
            local_file_path = os.path.join(root, file)
            gcs_file_path = os.path.relpath(local_file_path, model_dir)
            blob = bucket.blob(os.path.join(new_model, gcs_file_path))  # Use new_model_name
            blob.upload_from_filename(local_file_path)

# Upload the fine-tuned model and tokenizer to GCS
upload_to_gcs(BUCKET_DATA_NAME, save_model_path)
print(f"Fine-tuned model {new_model} successfully uploaded to GCS.")

Dockerfile

# Using the NVIDIA CUDA base image
FROM nvidia/cuda:12.6.2-runtime-ubuntu22.04

# Install necessary system packages
RUN apt-get update && \
    apt-get -y --no-install-recommends install python3-dev gcc python3-pip git && \
    rm -rf /var/lib/apt/lists/*

# Copy requirements.txt into the container
COPY requirements.txt .

# Install Python packages from requirements.txt
RUN pip install --no-cache-dir --upgrade pip && pip install --no-cache-dir -r requirements.txt

# Copy your finetune script into the container
COPY finetuning.py .

# Set the environment variable to ensure output is flushed
ENV PYTHONUNBUFFERED 1
ENV MODEL_ID "google/gemma-2-9b-it"
ENV GCS_BUCKET "finetuning-data-bucket"
 
# Set the command to run the finetuning script with CUDA device
CMD ["python3", "finetuning.py"]

requirements.txt

accelerate==1.1.1
bitsandbytes==0.45.0
datasets==3.1.0
gcsfs==2024.9.0
peft==v0.13.2
torch==2.5.1
transformers==4.47.0
trl==v0.11.4

Nous allons maintenant créer une image de conteneur pour l'affinage et la transférer vers Artifact Registry.

gcloud builds submit --tag ${REGION}-docker.pkg.dev/${DEVSHELL_PROJECT_ID}/${REPO_NAME}/finetuning:latest

12. Présentation d'Airflow 2, y compris ce qu'est un DAG

Airflow est une plate-forme d'orchestration de workflows et de pipelines de données. Il utilise des DAG (graphes orientés acycliques) pour définir ces workflows dans du code Python, en représentant visuellement les tâches et leurs dépendances.

Airflow, avec ses DAG statiques et ses définitions basées sur Python, est bien adapté à la planification et à la gestion des workflows prédéfinis. Son architecture inclut une interface utilisateur conviviale pour surveiller et gérer ces workflows.

En substance, Airflow vous permet de définir, de planifier et de surveiller vos pipelines de données à l'aide de Python, ce qui en fait un outil flexible et puissant pour l'orchestration de workflows.

13. Présentation de notre DAG

DAG signifie "Directed Acyclic Graph" (graphe orienté acyclique). Dans Airflow, un DAG représente l'ensemble du workflow ou du pipeline. Il définit les tâches, leurs dépendances et l'ordre d'exécution.

Les unités de workflow au sein du DAG sont exécutées à partir d'un pod sur le cluster GKE, initié à partir de la configuration Airflow.

Récapitulatif :

Airflow : téléchargement de données : ce script automatise le processus d'obtention d'un ensemble de données d'avis sur des films à partir de Kaggle et de stockage dans votre bucket GCS. Il est ainsi facilement disponible pour un traitement ou une analyse ultérieurs dans votre environnement cloud.

Airflow : préparation des données : le code prend l'ensemble de données brutes des avis sur les films, supprime les colonnes de données superflues qui ne sont pas nécessaires pour notre cas d'utilisation et supprime les ensembles de données comportant des valeurs manquantes. Il structure ensuite l'ensemble de données dans un format de questions/réponses adapté au machine learning et le stocke dans GCS pour une utilisation ultérieure.

Airflow : affiner un modèle : ce code affine un grand modèle de langage (LLM) à l'aide d'une technique appelée LoRA (Low-Rank Adaptation) et enregistre ensuite le modèle mis à jour. Elle commence par charger un LLM pré-entraîné et un ensemble de données depuis Google Cloud Storage. Il applique ensuite LoRA pour affiner efficacement le modèle sur cet ensemble de données. Enfin, il enregistre le modèle affiné dans Google Cloud Storage pour une utilisation ultérieure dans des applications telles que la génération de texte ou la réponse aux questions.

Airflow : mise en service de modèles : mise en service du modèle affiné sur GKE avec vllm pour l'inférence.

Airflow : boucle de rétroaction : réentraînement du modèle toutes les xx heures (toutes les heures, tous les jours, toutes les semaines).

Ce diagramme explique comment fonctionne Airflow 2 lorsqu'il est exécuté sur GKE.

14. Affinage d'un modèle ou utilisation du RAG

Cet atelier de programmation affine un LLM au lieu d'utiliser la génération augmentée par récupération (RAG).

Comparons ces deux approches :

Affinage : crée un modèle spécialisé. L'affinage adapte le LLM à une tâche ou à un ensemble de données spécifiques, ce qui lui permet de fonctionner de manière autonome sans s'appuyer sur des sources de données externes.

Simplification de l'inférence : cela élimine le besoin d'un système et d'une base de données de récupération distincts, ce qui permet d'obtenir des réponses plus rapides et moins coûteuses, en particulier pour les cas d'utilisation fréquents.

RAG : s'appuie sur des connaissances externes. Le RAG récupère des informations pertinentes dans une base de connaissances pour chaque requête, ce qui garantit l'accès à des données spécifiques et à jour.

Complexité accrue : l'implémentation de la RAG dans un environnement de production tel qu'un cluster Kubernetes implique souvent plusieurs microservices pour le traitement et la récupération des données, ce qui peut augmenter la latence et les coûts de calcul.

Pourquoi avons-nous choisi le fine-tuning ?

Bien que la RAG soit adaptée au petit ensemble de données utilisé dans cet atelier de programmation, nous avons opté pour le réglage fin afin de présenter un cas d'utilisation typique d'Airflow. Ce choix nous permet de nous concentrer sur les aspects d'orchestration du workflow plutôt que de nous pencher sur les nuances de la configuration d'une infrastructure et de microservices supplémentaires pour la RAG.

Conclusion :

L'affinage et la RAG sont des techniques utiles qui présentent chacune des avantages et des inconvénients. Le choix optimal dépend des exigences spécifiques de votre projet, telles que la taille et la complexité de vos données, les besoins en termes de performances et les considérations de coûts.

15. Tâche de DAG n° 1 : Créer votre première étape sur Airflow : téléchargement de données

Pour vous donner un aperçu de cette unité DAG, notre code Python hébergé dans une image de conteneur télécharge le dernier ensemble de données RottenTomatoes depuis Kaggle.

Ne copiez pas ce code dans le bucket GCS. Nous copions mlops-dag.py en dernière étape, qui contient toutes les étapes de l'unité DAG dans un seul script Python.

mlops-dag.py

import yaml

from os import path
from datetime import datetime

from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

from kubernetes import client, config
from kubernetes.client import models
from kubernetes.client.rest import ApiException

GCP_PROJECT_ID = Variable.get("GCP_PROJECT_ID")
BUCKET_DATA_NAME = Variable.get("BUCKET_DATA_NAME")
HF_TOKEN = Variable.get("HF_TOKEN")
KAGGLE_USERNAME = Variable.get("KAGGLE_USERNAME")
KAGGLE_KEY = Variable.get("KAGGLE_KEY")
JOB_NAMESPACE = Variable.get("JOB_NAMESPACE", default_var="airflow")

with DAG(dag_id="mlops-dag",
            start_date=datetime(2024,11,1),
            schedule_interval="@daily",
            catchup=False) as dag:

        # Step 1: Fetch raw data to GCS Bucket
        dataset_download = KubernetesPodOperator(
            task_id="dataset_download_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/dataset-download:latest",
            name="dataset-download",
            service_account_name="airflow-mlops-sa",
            env_vars={
                    "KAGGLE_USERNAME":KAGGLE_USERNAME,
                    "KAGGLE_KEY":KAGGLE_KEY,
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME
            }
        )

        dataset_download

16. Tâche de DAG n° 2 : Créez votre deuxième étape sur Airflow : la préparation des données

Pour résumer cette unité DAG, nous chargeons un fichier CSV (rotten_tomatoes_movie_reviews.csv) depuis GCS dans un DataFrame Pandas.

Ensuite, nous limitons le nombre de lignes traitées à l'aide de DATASET_LIMIT pour les tests et l'efficacité des ressources, puis nous convertissons les données transformées en un ensemble de données Hugging Face.

Si vous regardez attentivement, vous verrez que nous entraînons 1 000 lignes dans le modèle avec "DATASET_LIMIT": "1000", car cela prend 20 minutes sur un GPU Nvidia L4.

Ne copiez pas ce code dans le bucket GCS. Nous copions mlops-dag.py à la dernière étape, qui contient toutes les étapes dans un seul script Python.

import yaml

from os import path
from datetime import datetime

from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

from kubernetes import client, config
from kubernetes.client import models
from kubernetes.client.rest import ApiException

GCP_PROJECT_ID = Variable.get("GCP_PROJECT_ID")
BUCKET_DATA_NAME = Variable.get("BUCKET_DATA_NAME")
HF_TOKEN = Variable.get("HF_TOKEN")
KAGGLE_USERNAME = Variable.get("KAGGLE_USERNAME")
KAGGLE_KEY = Variable.get("KAGGLE_KEY")
JOB_NAMESPACE = Variable.get("JOB_NAMESPACE", default_var="airflow")

with DAG(dag_id="mlops-dag",
            start_date=datetime(2024,11,1),
            schedule_interval="@daily",
            catchup=False) as dag:

        # Step 1: Fetch raw data to GCS Bucket
        dataset_download = KubernetesPodOperator(
            task_id="dataset_download_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/dataset-download:latest",
            name="dataset-download",
            service_account_name="airflow-mlops-sa",
            env_vars={
                    "KAGGLE_USERNAME":KAGGLE_USERNAME,
                    "KAGGLE_KEY":KAGGLE_KEY,
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME
            }
        )

        # Step 2: Run GKEJob for data preparation
        data_preparation = KubernetesPodOperator(
            task_id="data_pipeline_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/data-preparation:latest",
            name="data-preparation",
            service_account_name="airflow-mlops-sa",
            env_vars={
                    "GCP_PROJECT_ID":GCP_PROJECT_ID,
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME,
                    "DATASET_LIMIT": "1000",
                    "HF_TOKEN":HF_TOKEN
            }
        )

        dataset_download >> data_preparation

17. Tâche 3 du DAG : créer votre troisième étape sur Airflow (affinage du modèle)

Pour résumer cette unité de DAG, nous exécutons finetune.py pour affiner le modèle Gemma avec notre nouvel ensemble de données.

Ne copiez pas ce code dans le bucket GCS. Nous copions mlops-dag.py à la dernière étape, qui contient toutes les étapes dans un seul script Python.

mlops-dag.py

import yaml

from os import path
from datetime import datetime

from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

from kubernetes import client, config
from kubernetes.client import models
from kubernetes.client.rest import ApiException

GCP_PROJECT_ID = Variable.get("GCP_PROJECT_ID")
BUCKET_DATA_NAME = Variable.get("BUCKET_DATA_NAME")
HF_TOKEN = Variable.get("HF_TOKEN")
KAGGLE_USERNAME = Variable.get("KAGGLE_USERNAME")
KAGGLE_KEY = Variable.get("KAGGLE_KEY")
JOB_NAMESPACE = Variable.get("JOB_NAMESPACE", default_var="airflow")

with DAG(dag_id="mlops-dag",
            start_date=datetime(2024,11,1),
            schedule_interval="@daily",
            catchup=False) as dag:

        # DAG Task 1: Fetch raw data to GCS Bucket
        dataset_download = KubernetesPodOperator(
            task_id="dataset_download_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/dataset-download:latest",
            name="dataset-download",
            service_account_name="airflow-mlops-sa",
            env_vars={
                    "KAGGLE_USERNAME":KAGGLE_USERNAME,
                    "KAGGLE_KEY":KAGGLE_KEY,
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME
            }
        )

        # DAG Task 2: Run GKEJob for data preparation
        data_preparation = KubernetesPodOperator(
            task_id="data_pipeline_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/data-preparation:latest",
            name="data-preparation",
            service_account_name="airflow-mlops-sa",
            env_vars={
                    "GCP_PROJECT_ID":GCP_PROJECT_ID,
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME,
                    "DATASET_LIMIT": "1000",
                    "HF_TOKEN":HF_TOKEN
            }
        )

        # DAG Task 3: Run GKEJob for fine tuning
        fine_tuning = KubernetesPodOperator(
            task_id="fine_tuning_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/finetuning:latest",
            name="fine-tuning",
            service_account_name="airflow-mlops-sa",
            startup_timeout_seconds=600,
            container_resources=models.V1ResourceRequirements(
                    requests={"nvidia.com/gpu": "1"},
                    limits={"nvidia.com/gpu": "1"}
            ),
            env_vars={
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME,
                    "HF_TOKEN":HF_TOKEN
            }
        )

        dataset_download >> data_preparation >> fine_tuning

18. Tâche n° 4 du DAG : créez votre dernière étape sur Airflow : inférence / mise en service du modèle

vLLM est une puissante bibliothèque Open Source spécialement conçue pour l'inférence hautes performances des LLM. Lorsqu'il est déployé sur Google Kubernetes Engine (GKE), il exploite l'évolutivité et l'efficacité de Kubernetes pour diffuser efficacement des LLM.

Résumé des étapes :

• Importez le DAG "mlops-dag.py" dans le bucket GCS.
• Copiez deux fichiers de configuration YAML Kubernetes pour configurer l'inférence dans un bucket GCS.

mlops-dag.py

import yaml

from os import path
from datetime import datetime

from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

from kubernetes import client, config
from kubernetes.client import models
from kubernetes.client.rest import ApiException

GCP_PROJECT_ID = Variable.get("GCP_PROJECT_ID")
BUCKET_DATA_NAME = Variable.get("BUCKET_DATA_NAME")
HF_TOKEN = Variable.get("HF_TOKEN")
KAGGLE_USERNAME = Variable.get("KAGGLE_USERNAME")
KAGGLE_KEY = Variable.get("KAGGLE_KEY")
JOB_NAMESPACE = Variable.get("JOB_NAMESPACE", default_var="airflow")

def model_serving():
    config.load_incluster_config()
    k8s_apps_v1 = client.AppsV1Api()
    k8s_core_v1 = client.CoreV1Api()

    while True:
        try:
            k8s_apps_v1.delete_namespaced_deployment(
                    namespace="airflow",
                    name="inference-deployment",
                    body=client.V1DeleteOptions(
                    propagation_policy="Foreground", grace_period_seconds=5
                    )
            )
        except ApiException:
            break
    print("Deployment inference-deployment deleted")
    
    with open(path.join(path.dirname(__file__), "inference.yaml")) as f:
        dep = yaml.safe_load(f)
        resp = k8s_apps_v1.create_namespaced_deployment(
            body=dep, namespace="airflow")
        print(f"Deployment created. Status='{resp.metadata.name}'")
    
    while True:
        try:
            k8s_core_v1.delete_namespaced_service(
                    namespace="airflow",
                    name="llm-service",
                    body=client.V1DeleteOptions(
                    propagation_policy="Foreground", grace_period_seconds=5
                    )
            )
        except ApiException:
            break
    print("Service llm-service deleted")

    with open(path.join(path.dirname(__file__), "inference-service.yaml")) as f:
        dep = yaml.safe_load(f)
        resp = k8s_core_v1.create_namespaced_service(
            body=dep, namespace="airflow")
        print(f"Service created. Status='{resp.metadata.name}'")

with DAG(dag_id="mlops-dag",
            start_date=datetime(2024,11,1),
            schedule_interval="@daily",
            catchup=False) as dag:

        # DAG Step 1: Fetch raw data to GCS Bucket
        dataset_download = KubernetesPodOperator(
            task_id="dataset_download_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/dataset-download:latest",
            name="dataset-download",
            service_account_name="airflow-mlops-sa",
            env_vars={
                    "KAGGLE_USERNAME":KAGGLE_USERNAME,
                    "KAGGLE_KEY":KAGGLE_KEY,
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME
            }
        )

        # DAG Step 2: Run GKEJob for data preparation
        data_preparation = KubernetesPodOperator(
            task_id="data_pipeline_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/data-preparation:latest",
            name="data-preparation",
            service_account_name="airflow-mlops-sa",
            env_vars={
                    "GCP_PROJECT_ID":GCP_PROJECT_ID,
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME,
                    "DATASET_LIMIT": "1000",
                    "HF_TOKEN":HF_TOKEN
            }
        )

        # DAG Step 3: Run GKEJob for fine tuning
        fine_tuning = KubernetesPodOperator(
            task_id="fine_tuning_task",
            namespace=JOB_NAMESPACE,
            image="us-central1-docker.pkg.dev/{{ var.value.GCP_PROJECT_ID }}/mlops-airflow-repo/finetuning:latest",
            name="fine-tuning",
            service_account_name="airflow-mlops-sa",
            startup_timeout_seconds=600,
            container_resources=models.V1ResourceRequirements(
                    requests={"nvidia.com/gpu": "1"},
                    limits={"nvidia.com/gpu": "1"}
            ),
            env_vars={
                    "BUCKET_DATA_NAME":BUCKET_DATA_NAME,
                    "HF_TOKEN":HF_TOKEN
            }
        )

        # DAG Step 4: Run GKE Deployment for model serving
        model_serving = PythonOperator(
            task_id="model_serving",
            python_callable=model_serving
        )

        dataset_download >> data_preparation >> fine_tuning >> model_serving

Importez votre script Python (fichier DAG) ainsi que les fichiers manifestes Kubernetes dans le bucket DAGS GCS.

gcloud storage cp mlops-dag.py gs://${BUCKET_DAGS_NAME}
gcloud storage cp manifests/inference.yaml gs://${BUCKET_DAGS_NAME}
gcloud storage cp manifests/inference-service.yaml gs://${BUCKET_DAGS_NAME}

Dans l'UI Airflow, vous verrez mlops-dag.

1. Sélectionnez "Réactiver".
2. Sélectionnez "Déclencher le DAG" pour effectuer un cycle MLOps manuel.

Une fois votre DAG terminé, un résultat semblable à celui-ci s'affiche dans l'UI Airflow.

Après la dernière étape, vous pouvez récupérer le point de terminaison du modèle et envoyer un prompt pour le tester.

Attendez environ cinq minutes avant d'exécuter la commande curl, afin que l'inférence du modèle puisse commencer et que l'équilibreur de charge puisse attribuer une adresse IP externe.

export MODEL_ENDPOINT=$(kubectl -n airflow get svc/llm-service --output jsonpath='{.status.loadBalancer.ingress[0].ip}')

curl -X POST http://${MODEL_ENDPOINT}:8000/generate -H "Content-Type: application/json" -d @- <<EOF
{
    "prompt": "Question: Review analysis for movie 'dangerous_men_2015'",
    "temperature": 0.1,
    "top_p": 1.0,
    "max_tokens": 128
}
EOF

Résultat :

{"text":["Question: Review analysis for movie ‘dangerous_men_2015'\nContext: The film is a mess, but it's a mess that's occasionally entertaining.\nAnswer: POSITIVE\nReason: While the film is a mess, it's a mess that's occasionally entertaining.\n}

19. Félicitations !

Vous avez créé votre premier workflow d'IA à l'aide d'un pipeline DAG avec Airflow 2 sur GKE.

N'oubliez pas de supprimer le provisionnement des ressources que vous avez déployées.

20. Effectuer cette opération en production

Cet atelier de programmation vous a permis de découvrir comment configurer Airflow 2 sur GKE. Toutefois, en conditions réelles, vous devrez tenir compte de certains des points suivants lorsque vous effectuerez cette opération en production.

Implémentez une interface Web à l'aide de Gradio ou d'un outil similaire.

Configurez la surveillance automatique des applications pour les charges de travail avec GKE ici ou exportez les métriques depuis Airflow ici.

Vous aurez peut-être besoin de GPU plus grands pour affiner le modèle plus rapidement, en particulier si vous disposez d'ensembles de données plus volumineux. Toutefois, si nous voulons entraîner le modèle sur plusieurs GPU, nous devons fractionner l'ensemble de données et répartir l'entraînement. Voici une explication de FSDP avec PyTorch (parallélisme des données entièrement segmentées, utilisant le partage de GPU pour atteindre cet objectif). Pour en savoir plus, consultez l'article de blog de Meta et ce tutoriel sur FSDP avec PyTorch.

Google Cloud Composer est un service Airflow géré. Vous n'avez donc pas besoin de gérer Airflow lui-même. Il vous suffit de déployer votre DAG et le tour est joué.

En savoir plus

• Documentation Airflow : https://airflow.apache.org/

Licence

Ce document est publié sous une licence Creative Commons Attribution 2.0 Generic.