Qualité des données programmatique avec Dataplex et l'IA générative

1. Introduction

Cet atelier de programmation fournit un plan technique aux spécialistes des données. Il décrit une approche "code-first" de la gouvernance des données, en montrant comment intégrer une gestion robuste de la qualité et des métadonnées directement dans le cycle de vie du développement. Au cœur de son fonctionnement, Dataplex Universal Catalog agit comme une data fabric intelligente, permettant aux entreprises de gérer, de surveiller et de gouverner de manière centralisée les données de l'ensemble de leur patrimoine, des lacs de données aux entrepôts.

L'atelier de programmation montre comment utiliser Dataplex, BigQuery et l'interface de ligne de commande Gemini pour aplatir des données complexes, les profiler de manière programmatique, générer des suggestions de règles de qualité des données intelligentes et déployer des analyses de qualité automatisées. L'objectif principal est de dépasser les processus manuels basés sur l'UI, qui sont sujets aux erreurs et difficiles à mettre à l'échelle, et d'établir à la place un framework "policy-as-code" robuste et contrôlable par version.

Prérequis

  • Connaissances de base concernant la console Google Cloud
  • Compétences de base concernant l'interface de ligne de commande et Google Cloud Shell

Points abordés

  • Découvrez comment aplatir les données BigQuery imbriquées à l'aide de vues matérialisées pour permettre un profilage complet.
  • Découvrez comment déclencher et gérer des analyses de profil Dataplex de manière programmatique à l'aide de la bibliothèque cliente Python Dataplex.
  • Comment exporter les données de profil et les structurer en tant qu'entrée pour un modèle d'IA générative.
  • Comment concevoir un prompt pour que la CLI Gemini analyse les données de profil et génère un fichier de règles YAML conforme à Dataplex.
  • Importance d'un processus interactif avec intervention humaine (HITL) pour valider les configurations générées par l'IA.
  • Déployer les règles générées en tant qu'analyse automatisée de la qualité des données

Prérequis

  • Un compte Google Cloud et un projet Google Cloud
  • Un navigateur Web tel que Chrome

Concepts clés : les piliers de la qualité des données Dataplex

Il est essentiel de comprendre les composants de base de Dataplex pour élaborer une stratégie efficace de qualité des données.

  • Analyse du profil de données : job Dataplex qui analyse les données et génère des métadonnées statistiques, y compris les pourcentages de valeurs nulles, le nombre de valeurs distinctes et les distributions de valeurs. Il s'agit de notre phase de "découverte" programmatique.
  • Règles sur la qualité des données : déclarations définissant les conditions que vos données doivent remplir (par exemple, NonNullExpectation, SetExpectation, RangeExpectation).
  • IA générative pour la suggestion de règles : utilisation d'un grand modèle de langage (comme Gemini) pour analyser un profil de données et suggérer des règles de qualité des données pertinentes. Cela accélère le processus de définition d'un framework de qualité de référence.
  • Analyse de la qualité des données : job Dataplex qui valide les données par rapport à un ensemble de règles prédéfinies ou personnalisées.
  • Gouvernance programmatique : thème central de la gestion des contrôles de gouvernance (comme les règles de qualité) en tant que code (par exemple, dans les fichiers YAML et les scripts Python). Cela permet l'automatisation, la gestion des versions et l'intégration dans les pipelines CI/CD.
  • Human-in-the-Loop (HITL) : point de contrôle essentiel pour intégrer l'expertise et la supervision humaines dans un workflow automatisé. Pour les configurations générées par l'IA, le processus HITL est essentiel pour valider l'exactitude, la pertinence commerciale et la sécurité des suggestions avant leur déploiement.

2. Préparation

Démarrer Cloud Shell

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

Dans la console Google Cloud, cliquez sur l'icône Cloud Shell dans la barre d'outils supérieure :

55efc1aaa7a4d3ad.png

Le provisionnement et la connexion à l'environnement prennent quelques instants seulement. Une fois l'opération terminée, le résultat devrait ressembler à ceci :

7ffe5cbb04455448.png

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 effectuer toutes les tâches de cet atelier de programmation dans un navigateur. Vous n'avez rien à installer.

Activer les API requises et configurer l'environnement

Dans Cloud Shell, assurez-vous que l'ID de votre projet est configuré :

export PROJECT_ID=$(gcloud config get-value project)
gcloud config set project $PROJECT_ID
export LOCATION="us-central1"
export BQ_LOCATION="us"
export DATASET_ID="dataplex_dq_codelab"
export TABLE_ID="ga4_transactions"

gcloud services enable dataplex.googleapis.com \
                       bigquery.googleapis.com \
                       serviceusage.googleapis.com

Dans l'exemple que nous utilisons, l'emplacement est us (multirégional), car les exemples de données publiques que nous allons utiliser se trouvent également dans us (multirégional). BigQuery exige que les données sources et la table de destination d'une requête se trouvent au même emplacement.

Créer un ensemble de données BigQuery dédié

Créez un ensemble de données BigQuery pour héberger nos exemples de données et nos résultats.

bq --location=us mk --dataset $PROJECT_ID:$DATASET_ID

Préparer les exemples de données

Pour cet atelier de programmation, vous allez utiliser un ensemble de données public contenant des données d'e-commerce obscurcies provenant du Google Merchandise Store. Étant donné que les ensembles de données publics sont en lecture seule, vous devez créer une copie modifiable dans votre propre ensemble de données. La commande bq suivante crée une table ga4_transactions dans votre ensemble de données dataplex_dq_codelab. Il copie les données d'une seule journée (2021-01-31) pour s'assurer que les analyses s'exécutent rapidement.

bq query \
--use_legacy_sql=false \
--destination_table=$PROJECT_ID:$DATASET_ID.$TABLE_ID \
--replace=true \
'SELECT * FROM `bigquery-public-data.ga4_obfuscated_sample_ecommerce.events_20210131`'

Configurer le répertoire de démonstration

Pour commencer, clonez un dépôt GitHub qui contient la structure de dossier et les fichiers d'assistance nécessaires pour cet atelier de programmation.

git clone https://github.com/GoogleCloudPlatform/devrel-demos
cd devrel-demos/data-analytics/programmatic-dq

Ce répertoire est désormais votre zone de travail active. Tous les fichiers suivants seront créés ici.

3. Découverte automatisée des données avec le profilage Dataplex

Le profilage des données Dataplex est un outil puissant qui permet de découvrir automatiquement des informations statistiques sur vos données, telles que les pourcentages de valeurs nulles, l'unicité et les distributions de valeurs. Ce processus est essentiel pour comprendre la structure et la qualité de vos données. Toutefois, une limitation connue du profilage Dataplex est son incapacité à inspecter entièrement les champs imbriqués ou répétés (par exemple, de type RECORD ou ARRAY) dans un tableau. Il peut identifier qu'une colonne est un type complexe, mais ne peut pas profiler les champs individuels de cette structure imbriquée.

Pour surmonter ce problème, nous allons aplatir les données dans des vues matérialisées spécialement conçues. Cette stratégie transforme chaque champ en colonne de premier niveau, ce qui permet à Dataplex de profiler chacun d'eux individuellement.

Comprendre le schéma imbriqué

Commençons par examiner le schéma de notre table source. L'ensemble de données Google Analytics 4 (GA4) contient plusieurs colonnes imbriquées et répétées. Pour récupérer le schéma complet de manière programmatique, y compris toutes les structures imbriquées, vous pouvez utiliser la commande bq show et enregistrer le résultat dans un fichier JSON.

bq show --schema --format=json $PROJECT_ID:$DATASET_ID.$TABLE_ID > bq_schema.json

L'inspection du fichier bq_schema.json révèle des structures complexes telles que des éléments d'appareil, de zone géographique, d'e-commerce et d'enregistrements répétés. Il s'agit des structures qui doivent être aplaties pour un profilage efficace.

Aplatir les données avec les vues matérialisées

La création de vues matérialisées est la solution la plus efficace et la plus pratique pour relever ce défi lié aux données imbriquées. En précalculant les résultats aplatis, les vues matérialisées offrent des avantages considérables en termes de performances et de coûts des requêtes, tout en fournissant une structure plus simple, de type relationnel, pour les analystes et les outils de profilage.

La première idée qui vient à l'esprit est peut-être d'aplatir le tout en une seule vue géante. Toutefois, cette approche intuitive cache un piège dangereux qui peut entraîner une grave corruption des données. Découvrons pourquoi il s'agit d'une erreur critique.

  1. mv_ga4_user_session_flat.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_user_session_flat`
OPTIONS (
  enable_refresh = true,
  refresh_interval_minutes = 30
) AS
SELECT
  event_date, event_timestamp, event_name, user_pseudo_id, user_id, stream_id, platform,
  device.category AS device_category,
  device.operating_system AS device_os,
  device.operating_system_version AS device_os_version,
  device.language AS device_language,
  device.web_info.browser AS device_browser,
  geo.continent AS geo_continent,
  geo.country AS geo_country,
  geo.region AS geo_region,
  geo.city AS geo_city,
  traffic_source.name AS traffic_source_name,
  traffic_source.medium AS traffic_source_medium,
  traffic_source.source AS traffic_source_source
FROM
  `$PROJECT_ID.$DATASET_ID.ga4_transactions`;
  1. mv_ga4_ecommerce_transactions.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_ecommerce_transactions`
OPTIONS (
  enable_refresh = true,
  refresh_interval_minutes = 30
) AS
SELECT
  event_date, event_timestamp, user_pseudo_id, ecommerce.transaction_id,
  ecommerce.total_item_quantity,
  ecommerce.purchase_revenue_in_usd,
  ecommerce.purchase_revenue,
  ecommerce.refund_value_in_usd,
  ecommerce.refund_value,
  ecommerce.shipping_value_in_usd,
  ecommerce.shipping_value,
  ecommerce.tax_value_in_usd,
  ecommerce.tax_value,
  ecommerce.unique_items
FROM
  `$PROJECT_ID.$DATASET_ID.ga4_transactions`
WHERE
  ecommerce.transaction_id IS NOT NULL;
  1. mv_ga4_ecommerce_items.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_ecommerce_items`
OPTIONS (
  enable_refresh = true,
  refresh_interval_minutes = 30
) AS
SELECT
  event_date, event_timestamp, event_name, user_pseudo_id, ecommerce.transaction_id,
  item.item_id,
  item.item_name,
  item.item_brand,
  item.item_variant,
  item.item_category,
  item.item_category2,
  item.item_category3,
  item.item_category4,
  item.item_category5,
  item.price_in_usd,
  item.price,
  item.quantity,
  item.item_revenue_in_usd,
  item.item_revenue,
  item.coupon,
  item.affiliation,
  item.item_list_name,
  item.promotion_name
FROM
  `$PROJECT_ID.$DATASET_ID.ga4_transactions`,
  UNNEST(items) AS item
WHERE
  ecommerce.transaction_id IS NOT NULL;

Exécutez maintenant ces modèles à l'aide de l'outil de ligne de commande bq. La commande envsubst lira chaque fichier, remplacera les variables telles que $PROJECT_ID et $DATASET_ID par leurs valeurs issues de votre environnement shell, puis transmettra le code SQL final et valide à la commande de requête bq.

envsubst < mv_ga4_user_session_flat.sql | bq query --use_legacy_sql=false
envsubst < mv_ga4_ecommerce_transactions.sql | bq query --use_legacy_sql=false
envsubst < mv_ga4_ecommerce_items.sql | bq query --use_legacy_sql=false

Exécuter les analyses de profil via le client Python

Maintenant que nous avons nos vues profilables et aplaties, nous pouvons créer et exécuter des analyses de profil de données Dataplex pour chacune d'elles de manière programmatique. Le script Python suivant utilise la bibliothèque cliente google-cloud-dataplex pour automatiser ce processus.

Avant d'exécuter le script, il est essentiel de créer un environnement Python isolé dans le répertoire de votre projet. Cela garantit que les dépendances du projet sont gérées séparément, ce qui évite les conflits avec d'autres packages dans votre environnement Cloud Shell.

# Create the virtual environment
python3 -m venv dq_venv

# Activate the environment
source dq_venv/bin/activate

Installez ensuite la bibliothèque cliente Dataplex dans l'environnement nouvellement activé.

# Install the Dataplex client library
pip install google-cloud-dataplex

Une fois l'environnement configuré et la bibliothèque installée, vous êtes prêt à créer le script d'orchestration.

Dans la barre d'outils Cloud Shell, cliquez sur "Ouvrir l'éditeur". Créez un fichier nommé 1_run_dataplex_scans.py et collez-y le code Python suivant. Si vous clonez le dépôt GitHub, ce fichier se trouve déjà dans votre dossier.

Ce script crée une analyse pour chaque vue matérialisée (si elle n'existe pas déjà), l'exécute, puis interroge jusqu'à ce que toutes les tâches d'analyse soient terminées.

import os
import sys
import time
from google.cloud import dataplex_v1
from google.api_core.exceptions import AlreadyExists


def create_and_run_scan(
    client: dataplex_v1.DataScanServiceClient,
    project_id: str,
    location: str,
    data_scan_id: str,
    target_resource: str,
) -> dataplex_v1.DataScanJob | None:
    """
    Creates and runs a single data profile scan.
    Returns the executed Job object without waiting for completion.
    """
    parent = client.data_scan_path(project_id, location, data_scan_id).rsplit('/', 2)[0]
    scan_path = client.data_scan_path(project_id, location, data_scan_id)

    # 1. Create Data Scan (skips if it already exists)
    try:
        data_scan = dataplex_v1.DataScan()
        data_scan.data.resource = target_resource
        data_scan.data_profile_spec = dataplex_v1.DataProfileSpec()

        print(f"[INFO] Creating data scan '{data_scan_id}'...")
        client.create_data_scan(
            parent=parent,
            data_scan=data_scan,
            data_scan_id=data_scan_id
        ).result()  # Wait for creation to complete
        print(f"[SUCCESS] Data scan '{data_scan_id}' created.")
    except AlreadyExists:
        print(f"[INFO] Data scan '{data_scan_id}' already exists. Skipping creation.")
    except Exception as e:
        print(f"[ERROR] Error creating data scan '{data_scan_id}': {e}")
        return None

    # 2. Run Data Scan
    try:
        print(f"[INFO] Running data scan '{data_scan_id}'...")
        run_response = client.run_data_scan(name=scan_path)
        print(f"[SUCCESS] Job started for '{data_scan_id}'. Job ID: {run_response.job.name.split('/')[-1]}")
        return run_response.job
    except Exception as e:
        print(f"[ERROR] Error running data scan '{data_scan_id}': {e}")
        return None


def main():
    """Main execution function"""
    # --- Load configuration from environment variables ---
    PROJECT_ID = os.environ.get("PROJECT_ID")
    LOCATION = os.environ.get("LOCATION")
    DATASET_ID = os.environ.get("DATASET_ID")

    if not all([PROJECT_ID, LOCATION, DATASET_ID]):
        print("[ERROR] One or more required environment variables are not set.")
        print("Please ensure PROJECT_ID, LOCATION, and DATASET_ID are exported in your shell.")
        sys.exit(1)

    print(f"[INFO] Using Project: {PROJECT_ID}, Location: {LOCATION}, Dataset: {DATASET_ID}")

    # List of Materialized Views to profile
    TARGET_VIEWS = [
        "mv_ga4_user_session_flat",
        "mv_ga4_ecommerce_transactions",
        "mv_ga4_ecommerce_items"
    ]
    # ----------------------------------------------------

    client = dataplex_v1.DataScanServiceClient()
    running_jobs = []

    # 1. Create and run jobs for all target views
    print("\n--- Starting Data Profiling Job Creation and Execution ---")
    for view_name in TARGET_VIEWS:
        data_scan_id = f"profile-scan-{view_name.replace('_', '-')}"
        target_resource = f"//bigquery.googleapis.com/projects/{PROJECT_ID}/datasets/{DATASET_ID}/tables/{view_name}"

        job = create_and_run_scan(client, PROJECT_ID, LOCATION, data_scan_id, target_resource)
        if job:
            running_jobs.append(job)
    print("-------------------------------------------------------\n")

    if not running_jobs:
        print("[ERROR] No jobs were started. Exiting.")
        return

    # 2. Poll for all jobs to complete
    print("--- Monitoring job completion status (checking every 30 seconds) ---")
    completed_jobs = {}

    while running_jobs:
        jobs_to_poll_next = []

        print(f"\n[STATUS] Checking status for {len(running_jobs)} running jobs...")

        for job in running_jobs:
            job_id_short = job.name.split('/')[-1][:13] 
            try:
                updated_job = client.get_data_scan_job(name=job.name)
                state = updated_job.state

                if state in (dataplex_v1.DataScanJob.State.RUNNING, dataplex_v1.DataScanJob.State.PENDING, dataplex_v1.DataScanJob.State.CANCELING):
                    print(f"  - Job {job_id_short}... Status: {state.name}")
                    jobs_to_poll_next.append(updated_job) 
                else:
                    print(f"  - Job {job_id_short}... Status: {state.name} (Complete)")
                    completed_jobs[job.name] = updated_job

            except Exception as e:
                print(f"[ERROR] Could not check status for job {job_id_short}: {e}")

        running_jobs = jobs_to_poll_next

        if running_jobs:
            time.sleep(30)

    # 3. Print final results
    print("\n--------------------------------------------------")
    print("[SUCCESS] All data profiling jobs have completed.")
    print("\nFinal Job Status Summary:")
    for job_name, job in completed_jobs.items():
        job_id_short = job_name.split('/')[-1][:13]
        print(f"  - Job {job_id_short}: {job.state.name}")
        if job.state == dataplex_v1.DataScanJob.State.FAILED:
            print(f"    - Failure Message: {job.message}")

    print("\nNext step: Analyze the profile results and generate quality rules.")


if __name__ == "__main__":
    main()

Exécutez maintenant le script à partir de votre terminal Cloud Shell.

python 1_run_dataplex_scans.py

Le script va maintenant orchestrer le profilage de vos trois vues matérialisées et vous fournir des informations sur l'état en temps réel. Une fois cette étape terminée, vous disposerez d'un profil statistique riche et lisible par machine pour chaque vue. Vous serez alors prêt à passer à l'étape suivante de notre workflow : la génération de règles de qualité des données optimisées par l'IA.

Vous pouvez consulter les analyses de profil terminées dans la console Google Cloud.

  1. Dans le menu de navigation, accédez à "Catalogue universel Dataplex" et "Profil" dans la section "Gouvernance".

5acda859404968c.png

  1. Vous devriez voir les trois analyses de profil listées, ainsi que l'état du dernier job. Vous pouvez cliquer sur une analyse pour afficher les résultats détaillés.

8a09dae0ef485289.png

Du profil BigQuery à l'entrée prête pour l'IA

Les analyses de profil Dataplex ont bien été exécutées. Bien que les résultats soient disponibles dans l'API Dataplex, pour les utiliser comme entrée pour un modèle d'IA générative, nous devons les extraire dans un fichier local structuré.

Le script Python suivant, 2_dq_profile_save.py, recherche de manière programmatique le dernier job d'analyse de profil réussi pour notre vue mv_ga4_user_session_flat. Il récupère ensuite le résultat complet et détaillé du profil, et l'enregistre en tant que fichier JSON local nommé dq_profile_results.json. Ce fichier servira d'entrée directe pour notre analyse de l'IA lors de la prochaine étape.

Dans l'éditeur Cloud Shell, créez un fichier nommé 2_dq_profile_save.py et collez-y le code suivant. Comme à l'étape précédente, vous pouvez ignorer la création d'un fichier si vous avez cloné le dépôt.

import os
import sys
import json
from google.cloud import dataplex_v1
from google.api_core.exceptions import NotFound
from google.protobuf.json_format import MessageToDict

# --- Configuration ---
# The Materialized View to analyze is fixed for this step.
TARGET_VIEW = "mv_ga4_user_session_flat"
OUTPUT_FILENAME = "dq_profile_results.json"


def save_to_json_file(content: dict, filename: str):
    """Saves the given dictionary content to a JSON file."""
    try:
        with open(filename, "w", encoding="utf-8") as f:
            # Use indent=2 for a readable, "pretty-printed" JSON file.
            json.dump(content, f, indent=2, ensure_ascii=False)
        print(f"\n[SUCCESS] Profile results were saved to '{filename}'.")
    except (IOError, TypeError) as e:
        print(f"[ERROR] An error occurred while saving the file: {e}")


def get_latest_successful_job(
    client: dataplex_v1.DataScanServiceClient,
    project_id: str,
    location: str,
    data_scan_id: str
) -> dataplex_v1.DataScanJob | None:
    """Finds and returns the most recently succeeded job for a given data scan."""
    scan_path = client.data_scan_path(project_id, location, data_scan_id)
    print(f"\n[INFO] Looking for the latest successful job for scan '{data_scan_id}'...")

    try:
        # List all jobs for the specified scan, which are ordered most-recent first.
        jobs_pager = client.list_data_scan_jobs(parent=scan_path)

        # Iterate through jobs to find the first one that succeeded.
        for job in jobs_pager:
            if job.state == dataplex_v1.DataScanJob.State.SUCCEEDED:
                return job

        # If no successful job is found after checking all pages.
        return None
    except NotFound:
        print(f"[WARN] No scan history found for '{data_scan_id}'.")
        return None


def main():
    """Main execution function."""
    # --- Load configuration from environment variables ---
    PROJECT_ID = os.environ.get("PROJECT_ID")
    LOCATION = os.environ.get("LOCATION")

    if not all([PROJECT_ID, LOCATION]):
        print("[ERROR] Required environment variables PROJECT_ID or LOCATION are not set.")
        sys.exit(1)

    print(f"[INFO] Using Project: {PROJECT_ID}, Location: {LOCATION}")
    print(f"--- Starting Profile Retrieval for: {TARGET_VIEW} ---")

    # Construct the data_scan_id based on the target view name.
    data_scan_id = f"profile-scan-{TARGET_VIEW.replace('_', '-')}"

    # 1. Initialize Dataplex client and get the latest successful job.
    client = dataplex_v1.DataScanServiceClient()
    latest_job = get_latest_successful_job(client, PROJECT_ID, LOCATION, data_scan_id)

    if not latest_job:
        print(f"\n[ERROR] No successful job record was found for '{data_scan_id}'.")
        print("Please ensure the 'run_dataplex_scans.py' script has completed successfully.")
        return

    job_id_short = latest_job.name.split('/')[-1]
    print(f"[SUCCESS] Found the latest successful job: '{job_id_short}'.")

    # 2. Fetch the full, detailed profile result for the job.
    print(f"[INFO] Retrieving detailed profile results for job '{job_id_short}'...")
    try:
        request = dataplex_v1.GetDataScanJobRequest(
            name=latest_job.name,
            view=dataplex_v1.GetDataScanJobRequest.DataScanJobView.FULL,
        )
        job_with_full_results = client.get_data_scan_job(request=request)
    except Exception as e:
        print(f"[ERROR] Failed to retrieve detailed job results: {e}")
        return

    # 3. Convert the profile result to a dictionary and save it to a JSON file.
    if job_with_full_results.data_profile_result:
        profile_dict = MessageToDict(job_with_full_results.data_profile_result._pb)
        save_to_json_file(profile_dict, OUTPUT_FILENAME)
    else:
        print("[WARN] The job completed, but no data profile result was found within it.")

    print("\n[INFO] Script finished successfully.")


if __name__ == "__main__":
    main()

Exécutez maintenant le script depuis votre terminal :

python 2_dq_profile_save.py

Une fois ces opérations effectuées, votre répertoire doit comporter un nouveau fichier nommé dq_profile_results.json. Ce fichier contient les métadonnées statistiques riches et détaillées que nous utiliserons pour générer des règles de qualité des données. Si vous souhaitez vérifier le contenu de dq_profile_results.json, exécutez la commande suivante :

cat dq_profile_results.json

4. Générer des règles de qualité des données avec la CLI Gemini

Installer et configurer la CLI Gemini

Bien que vous puissiez appeler l'API Gemini de manière programmatique, l'utilisation d'un outil tel que la CLI Gemini offre un moyen puissant et interactif d'intégrer l'IA générative directement dans vos workflows de terminal. La CLI Gemini n'est pas qu'un chatbot. C'est un outil de workflow en ligne de commande qui peut lire vos fichiers locaux, comprendre votre code et interagir avec d'autres outils système, comme gcloud, pour automatiser des tâches complexes. C'est donc idéal pour notre cas d'utilisation.

Conditions préalables

Tout d'abord, assurez-vous de disposer du prérequis requis : Node.js version 20 ou ultérieure doit être installé dans votre environnement Cloud Shell. Vous pouvez vérifier votre version en exécutant node -v.

Installation

Il existe deux façons d'utiliser la CLI Gemini : l'installer temporairement ou de manière plus permanente. Nous aborderons les deux méthodes ici.

Vous pouvez exécuter la CLI Gemini directement pour une seule session, sans aucune installation permanente. Il s'agit du moyen le plus simple et le plus rapide de tester l'outil, car votre environnement reste complètement inchangé.

Dans votre terminal Cloud Shell, exécutez la commande suivante :

npx https://github.com/google-gemini/gemini-cli

Cette commande télécharge et exécute temporairement le package CLI.

Pour tout projet réel, la bonne pratique recommandée consiste à installer l'interface de ligne de commande localement dans le répertoire de votre projet. Cette approche présente plusieurs avantages clés :

  • Isolation des dépendances : elle garantit que votre projet dispose de sa propre version de la CLI, ce qui évite les conflits de version avec d'autres projets.
  • Reproductibilité : toute personne qui clone votre projet peut installer exactement les mêmes dépendances, ce qui rend votre configuration fiable et portable.
  • Alignement sur les bonnes pratiques : il suit le modèle standard de gestion des dépendances des projets Node.js, en évitant les pièges des installations globales (-g).

Pour installer la CLI en local, exécutez la commande suivante à partir du dossier de votre projet (programmatic-dq) :

npm install @google/gemini-cli

Cela crée un dossier node_modules dans programmatic-dq. Pour exécuter la version que vous venez d'installer, utilisez la commande npx.

npx gemini

Première configuration

Quelle que soit la méthode choisie, la première fois que vous lancerez la CLI, vous serez guidé tout au long d'une procédure de configuration unique.

8a25fab5951c6c39.png

Vous serez invité à choisir un thème de couleur, puis à vous authentifier. La méthode la plus simple consiste à vous connecter avec votre compte Google lorsque vous y êtes invité. Le niveau sans frais fourni est suffisant pour cet atelier de programmation.

Maintenant que l'interface de ligne de commande est installée et configurée, vous pouvez générer les règles. L'interface CLI connaît les fichiers de son répertoire actuel, ce qui est essentiel pour l'étape suivante.

Générer les règles de qualité des données

Bien qu'il soit possible de demander à un LLM de générer un fichier de configuration en une seule fois, la nature non déterministe des modèles génératifs signifie que le résultat ne sera pas toujours parfaitement conforme au schéma strict requis par des outils tels que gcloud. Une méthode plus fiable consiste en un processus interactif en plusieurs étapes. L'IA agit d'abord en tant qu'analyste pour proposer un plan. Vous (l'expert humain) examinez et approuvez ce plan. Ce n'est qu'ensuite que l'IA génère le code final en fonction de vos instructions approuvées.

Cette approche transforme le processus, qui passe d'une simple commande à une session collaborative. Elle garantit que le résultat final est précis, validé et prêt à être déployé.

Nous allons d'abord demander à Gemini d'analyser le profil statistique et d'agir en tant qu'ingénieur de données expert en proposant un plan d'action. Nous lui indiquons explicitement de ne pas écrire le fichier YAML pour le moment. Il se concentre sur l'analyse et la justification.

Dans votre session interactive Gemini CLI, collez le prompt suivant :

cd9792a6f1eb9f3.png

You are an expert Google Cloud Dataplex engineer. 
Your first task is to create a plan. I have a file in the current directory named ./dq_profile_results.json. 
Based on the statistical data within that file, propose a step-by-step plan to create a Dataplex data quality rules file. 
Your plan should identify which specific columns are good candidates for rules like nonNullExpectation, setExpectation, or rangeExpectation, and explain why based on the metrics (e.g., "Plan to create a nonNullExpectation for column X because its null percentage is 0%"). 
Do not write the YAML file yet. Just provide the plan.

Gemini analysera le fichier JSON et devrait renvoyer un plan structuré semblable à celui-ci :

Plan to create Dataplex data quality rules

1. Create a row count rule:
  - Analysis: The profile shows a total of 26,489 rows.
  - Proposed Rule: I will implement a rowCount rule to ensure the table's row count remains within a healthy range (e.g., 20,000 to 30,000). This is a fundamental check to guard against major data loading failures or unexpected deletions.
2. Implement non-null expectation rules:
  - Analysis: The profiling results show that many columns have no null values. For example, event_date, event_timestamp, event_name, user_pseudo_id, stream_id, and platform all appear to be fully populated.
  - Proposed Rules: I will create nonNullExpectation rules for columns where the nullRatio is 0 or not present, indicating they are always expected to contain a value:
    - event_date
    - event_timestamp
    - event_name
    - user_pseudo_id
    - stream_id
    - platform
    - device_category
    - device_os
3. Implement set expectation rules:
  - Analysis: Several columns contain a small, finite set of values, making them ideal for set validation.
  - Proposed Rules: I will create setExpectation rules for these columns:
    - platform: It only contains the value "WEB".
    - device_category: It contains only "desktop", "mobile", and "tablet".
    - geo_continent: It has a well-defined list of continents.
traffic_source_medium: It has a small list of valid mediums like "organic", "(none)", and "referral".
4. Implement range expectation rules:
  - Analysis: The profile provides minimum and maximum values for numeric and string length fields.
  - Proposed Rules: I will create rangeExpectation rules for the following:
    - event_timestamp: The profile shows a clear min and max. I will use these values to create a range check.
    - event_date (String Length): The profile indicates that the minLength and maxLength are both 8, which is perfect for enforcing a YYYYMMDD format.

Il s'agit de l'étape la plus importante de l'ensemble du workflow : l'examen avec intervention humaine (HITL). Le plan généré par Gemini est basé uniquement sur des modèles statistiques dans les données. Il ne comprend pas le contexte de votre activité, les futures modifications de données ni l'intention spécifique derrière vos données. Votre rôle en tant qu'expert humain consiste à valider, corriger et approuver ce plan avant de le transformer en code.

Examinez attentivement le plan fourni par Gemini.

  • Est-ce clair ?
  • Est-il conforme à vos connaissances de l'entreprise ?
  • Existe-t-il des règles statistiquement valables, mais pratiquement inutiles ?

Les résultats que vous recevez de Gemini peuvent varier. Votre objectif est de l'affiner.

Par exemple, imaginons que le plan suggère une règle rowCount, car le tableau comporte un nombre fixe de lignes dans les données d'échantillon. En tant qu'expert humain, vous savez peut-être que la taille de cette table devrait augmenter chaque jour. Une règle stricte sur le nombre de lignes n'est donc pas pratique et risque de générer de fausses alertes. Il s'agit d'un exemple parfait d'application du contexte métier qui manque à l'IA.

Vous allez maintenant fournir des commentaires à Gemini et lui donner la commande finale pour générer le code. Vous devez adapter la requête suivante en fonction du plan que vous avez réellement reçu et des corrections que vous souhaitez apporter.

La requête ci-dessous est un modèle. La première ligne vous permet de fournir vos corrections spécifiques. Si le plan que Gemini vous a fourni est parfait et n'a pas besoin d'être modifié, vous pouvez simplement supprimer cette ligne.

Dans la même session Gemini, collez votre version adaptée de la requête suivante :

[YOUR CORRECTIONS AND APPROVAL GO HERE. Examples:
- "The plan looks good. Please proceed."
- "The rowCount rule is not necessary, as the table size changes daily. The rest of the plan is approved. Please proceed."
- "For the setExpectation on the geo_continent column, please also include 'Antarctica'."]

Once you have incorporated my feedback, please generate the `dq_rules.yaml` file.

You must adhere to the following strict requirements:

- Schema Compliance: The YAML structure must strictly conform to the DataQualityRule specification. For a definitive source of truth, you must refer to the sample_rule.yaml file in the current directory and the DataQualityRule class definition in the local virtual environment path: ./dq_venv/.../google/cloud/dataplex_v1/types/data_quality.py.

- Data-Driven Values: All rule parameters, such as thresholds or expected values, must be derived directly from the statistical metrics in dq_profile_results.json.

- Rule Justification: For each rule, add a comment (#) on the line above explaining the justification, as you outlined in your plan.

- Output Purity: The final output must only be the raw YAML code block, perfectly formatted and ready for immediate deployment.

Gemini génère alors le contenu YAML en fonction de vos instructions précises et validées par un humain. Une fois l'opération terminée, un fichier dq_rules.yaml est créé dans votre répertoire de travail.

Créer et exécuter l'analyse de la qualité des données

Maintenant que vous disposez d'un fichier dq_rules.yaml généré par l'IA et validé par un humain, vous pouvez le déployer en toute confiance.

Quittez la CLI Gemini en saisissant /quit ou en appuyant deux fois sur Ctrl+C.

La commande gcloud suivante crée une ressource d'analyse de données Dataplex. Il n'exécute pas encore l'analyse. Il enregistre simplement la définition et la configuration de l'analyse (notre fichier YAML) auprès de Dataplex.

Exécutez la commande suivante dans votre terminal :

export DQ_SCAN="dq-scan"
gcloud dataplex datascans create data-quality $DQ_SCAN \
    --project=$PROJECT_ID \
    --location=$REGION \
    --data-quality-spec-file=dq_rules.yaml \
    --data-source-resource="//bigquery.googleapis.com/projects/$PROJECT_ID/datasets/$DATASET_ID/tables/mv_ga4_user_session_flat"

Maintenant que l'analyse est définie, vous pouvez déclencher un job pour l'exécuter.

gcloud dataplex datascans run $DQ_SCAN --location=$REGION --project=$PROJECT_ID

Cette commande génère un ID de job. Vous pouvez surveiller l'état de ce job dans la section Dataplex de la console Google Cloud. Une fois l'opération terminée, les résultats seront écrits dans une table BigQuery pour analyse.

5. Rôle essentiel de l'approche HITL (human-in-the-loop)

Bien que l'utilisation de Gemini pour accélérer la génération de règles soit extrêmement efficace, il est essentiel de considérer l'IA comme un copilote très compétent, et non comme un pilote entièrement autonome. Le processus HITL (human-in-the-loop, ou intervention humaine) n'est pas une suggestion facultative. Il s'agit d'une étape fondamentale et non négociable de tout workflow de gouvernance des données robuste et fiable. Le simple déploiement d'artefacts générés par l'IA sans une supervision humaine rigoureuse est voué à l'échec.

Considérez le dq_rules.yaml généré par l'IA comme une demande d'extraction envoyée par un développeur d'IA extrêmement rapide, mais inexpérimenté. Il doit être examiné en détail par un expert humain senior (vous) avant de pouvoir être fusionné dans la "branche principale" de votre règlement sur la gouvernance et déployé. Cet examen est essentiel pour atténuer les faiblesses inhérentes aux grands modèles de langage.

Voici une explication détaillée de l'importance de cet examen manuel et des éléments spécifiques que vous devez rechercher :

1. Validation contextuelle : l'IA manque de connaissances sur l'entreprise

  • La faiblesse des LLM : un LLM est un expert en modèles et en statistiques, mais il ne comprend absolument rien au contexte de votre entreprise. Par exemple, si une colonne new_campaign_id présente un taux de valeurs nulles de 98 %, un LLM peut ignorer cette colonne pour une raison statistique.
  • Le rôle essentiel de l'humain : vous, l'expert humain, savez que le champ new_campaign_id vient d'être ajouté hier pour un lancement de produit majeur la semaine prochaine. Vous savez que son ratio de valeurs nulles devrait être élevé maintenant, mais qu'il devrait chuter de manière significative. Vous savez également qu'une fois renseignée, elle doit respecter un format spécifique. L'IA ne peut pas inférer ces connaissances externes sur l'entreprise. Votre rôle consiste à appliquer ce contexte commercial aux suggestions statistiques de l'IA, en les remplaçant ou en les complétant si nécessaire.

2. Exactitude et précision : protection contre les hallucinations et les erreurs subtiles

  • Point faible des LLM : les LLM peuvent être "confiants dans leur erreur". Ils peuvent "halluciner" ou générer du code qui est subtilement incorrect. Par exemple, il peut générer un fichier YAML avec une règle correctement nommée, mais un paramètre non valide, ou il peut mal orthographier un type de règle (par exemple, setExpectations au lieu de setExpectation). Ces erreurs subtiles peuvent être difficiles à repérer, mais elles entraînent l'échec du déploiement.
  • Rôle essentiel de l'humain : votre rôle consiste à agir en tant que linter et validateur de schéma ultime. Vous devez vérifier méticuleusement le fichier YAML généré par rapport à la spécification DataQualityRule officielle de Dataplex. Vous ne vous contentez pas de vérifier si le code "semble correct". Vous validez sa correction syntaxique et sémantique pour vous assurer qu'il est 100 % conforme à l'API cible. C'est pourquoi l'atelier de programmation invite Gemini à référencer les fichiers de schéma pour réduire les risques d'erreur, mais la vérification finale vous incombe.

3. Sécurité et atténuation des risques : éviter les conséquences en aval

  • La faiblesse des LLM : une règle de qualité des données défectueuse déployée en production peut avoir de graves conséquences. Si l'IA suggère un rangeExpectation trop large pour un montant de transaction financière, elle risque de ne pas détecter d'activité frauduleuse. À l'inverse, si elle suggère une règle trop stricte basée sur un petit échantillon de données, elle pourrait inonder votre équipe d'astreinte de milliers d'alertes de faux positifs, entraînant une fatigue liée aux alertes et risquant de faire passer à côté de vrais problèmes.
  • Rôle essentiel de l'humain : vous êtes l'ingénieur en sécurité. Vous devez évaluer l'impact potentiel en aval de chaque règle suggérée par l'IA. Posez-vous la question suivante : "Que se passe-t-il si cette règle échoue ? L'alerte est-elle exploitable ? Quel est le risque si cette règle est validée à tort ?" Cette évaluation des risques est une capacité propre à l'humain, qui met en balance le coût de l'échec et l'intérêt de la vérification.

4. La gouvernance en tant que processus continu : intégrer des connaissances prospectives

  • La faiblesse du LLM : les connaissances de l'IA sont basées sur un instantané statique des données. Les résultats du profil proviennent d'un moment précis. Il n'a aucune connaissance des événements futurs.
  • Le rôle essentiel de l'humain : votre stratégie de gouvernance doit être tournée vers l'avenir. Vous savez qu'une source de données doit être migrée le mois prochain, ce qui modifiera le stream_id. Vous savez qu'un nouveau pays est ajouté à la liste geo_country. Le processus HITL consiste à injecter ces connaissances sur l'état futur, en mettant à jour ou en désactivant temporairement les règles pour éviter les problèmes lors des évolutions techniques ou commerciales prévues. La qualité des données n'est pas une configuration ponctuelle. Il s'agit d'un processus évolutif qui doit être guidé par un humain.

En résumé, HITL est le mécanisme essentiel de contrôle qualité et de sécurité qui transforme la gouvernance basée sur l'IA, qui était une idée nouvelle mais risquée, en une pratique responsable, évolutive et de niveau professionnel. Cela permet de s'assurer que les règles finales déployées ne sont pas seulement accélérées par l'IA, mais qu'elles sont également validées par des humains, combinant ainsi la rapidité des machines à la sagesse et au contexte des experts humains.

Toutefois, cette importance accordée à la supervision humaine ne diminue pas la valeur de l'IA. Au contraire, l'IA générative joue un rôle crucial dans l'accélération du processus HITL lui-même.

Sans IA, un ingénieur de données devrait :

  1. Écrivez manuellement des requêtes SQL complexes pour profiler les données (par exemple, COUNT DISTINCT, AVG, MIN, MAX pour chaque colonne).
  2. Analysez minutieusement les résultats feuille de calcul par feuille de calcul.
  3. Écrire chaque ligne du fichier de règles YAML à partir de zéro, une tâche fastidieuse et sujette aux erreurs.

L'IA automatise ces étapes laborieuses et chronophages. Il agit comme un analyste infatigable qui traite instantanément le profil statistique et fournit une "première version" bien structurée et complète à 80 % de la règle.

Cela modifie fondamentalement la nature du travail humain. Au lieu de passer des heures à traiter manuellement les données et à coder des modèles, l'expert humain peut se concentrer immédiatement sur les tâches les plus importantes :

  • Appliquer un contexte métier critique.
  • Valider l'exactitude de la logique de l'IA.
  • Prendre des décisions stratégiques sur les règles qui sont vraiment importantes.

Dans ce partenariat, l'IA s'occupe du "quoi" (quels sont les modèles statistiques ?), ce qui permet à l'humain de se concentrer sur le "pourquoi" (pourquoi ce modèle est-il important pour notre activité ?) et le "et alors ?" (quelle devrait être notre stratégie ?). L'IA ne remplace donc pas la boucle, mais rend chaque cycle plus rapide, plus intelligent et plus efficace.

6. Nettoyer votre environnement

Pour éviter que les ressources utilisées dans cet atelier de programmation ne soient facturées sur votre compte Google Cloud, supprimez le projet contenant les ressources. Toutefois, si vous souhaitez conserver le projet, vous pouvez supprimer les ressources individuelles que vous avez créées.

Supprimer les analyses Dataplex

Commencez par supprimer les analyses de profil et de qualité que vous avez créées. Pour éviter la suppression accidentelle de ressources importantes, ces commandes utilisent les noms spécifiques des analyses créées dans cet atelier de programmation.

# Delete the Data Quality Scan
gcloud dataplex datascans delete dq-scan \
    --location=us-central1 \
    --project=$PROJECT_ID --quiet

# Delete the Data Profile Scans
gcloud dataplex data-scans delete profile-scan-mv-ga4-user-session-flat \
    --location=us-central1 \
    --project=$PROJECT_ID --quiet

gcloud dataplex data-scans delete profile-scan-mv-ga4-ecommerce-transactions \
    --location=us-central1 \
    --project=$PROJECT_ID --quiet

gcloud dataplex data-scans delete profile-scan-mv-ga4-ecommerce-items \
    --location=us-central1 \
    --project=$PROJECT_ID --quiet

Supprimer l'ensemble de données BigQuery

Supprimez ensuite l'ensemble de données BigQuery. Cette commande est irréversible et utilise l'indicateur -f (force) pour supprimer l'ensemble de données et toutes ses tables sans confirmation.

# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:dataplex_dq_codelab

7. Félicitations !

Vous avez terminé cet atelier de programmation.

Vous avez créé un workflow de gouvernance des données programmatique de bout en bout. Vous avez commencé par utiliser des vues matérialisées pour simplifier des données BigQuery complexes et les rendre adaptées à l'analyse. Vous avez ensuite exécuté des analyses de profil Dataplex de manière programmatique pour générer des métadonnées statistiques. Plus important encore, vous avez utilisé la CLI Gemini pour analyser la sortie du profil et générer intelligemment un artefact "policy-as-code" (dq_rules.yaml). Vous avez ensuite utilisé la CLI pour déployer cette configuration en tant qu'analyse automatisée de la qualité des données, ce qui a permis de boucler la boucle sur une stratégie de gouvernance moderne et évolutive.

Vous disposez désormais du modèle fondamental pour créer des systèmes de qualité des données fiables, accélérés par l'IA et validés par des humains sur Google Cloud.

Et ensuite ?

  • Intégrer à CI/CD : prenez le fichier dq_rules.yaml et validez-le dans un dépôt Git. Créez un pipeline CI/CD (par exemple, à l'aide de Cloud Build ou de GitHub Actions) qui déploie automatiquement l'analyse Dataplex chaque fois que le fichier de règles est mis à jour.
  • Explorer les règles SQL personnalisées : allez au-delà des types de règles standards. Dataplex est compatible avec les règles SQL personnalisées pour appliquer une logique plus complexe et spécifique à l'entreprise, qui ne peut pas être exprimée avec des vérifications prédéfinies. Il s'agit d'une fonctionnalité puissante qui vous permet d'adapter la validation à vos besoins spécifiques.
  • Optimiser les analyses pour l'efficacité et le coût : pour les très grandes tables, vous pouvez améliorer les performances et réduire les coûts en n'analysant pas toujours l'ensemble de données. Explorez les filtres pour limiter l'analyse à des périodes ou des segments de données spécifiques, ou configurez des analyses échantillonnées pour vérifier un pourcentage représentatif de vos données.
  • Visualisez les résultats : la sortie de chaque analyse de la qualité des données Dataplex est écrite dans une table BigQuery. Associez cette table à Looker Studio pour créer des tableaux de bord qui suivent vos scores de qualité des données au fil du temps, agrégés selon les dimensions que vous avez définies (par exemple, Exhaustivité, Validité). La surveillance est ainsi proactive et visible pour toutes les parties prenantes.
  • Partagez les bonnes pratiques : encouragez le partage des connaissances au sein de votre organisation pour tirer parti de l'expérience collective et améliorer votre stratégie de qualité des données. Pour exploiter pleinement vos efforts de gouvernance, il est essentiel de favoriser une culture de confiance dans les données.
  • Consultez la documentation :