1. Wprowadzenie
To ćwiczenie zawiera techniczny plan działania dla specjalistów ds. danych. Opisuje ona podejście „najpierw kod” do zarządzania danymi, pokazując, jak włączyć solidne zarządzanie jakością i metadanymi bezpośrednio w cykl życia rozwoju. Knowledge Catalog to inteligentna tkanina danych, która umożliwia organizacjom centralne zarządzanie danymi w całym środowisku, od jezior danych po hurtownie, oraz monitorowanie i kontrolowanie tych danych.
W tym ćwiczeniu pokazujemy, jak wykorzystać Knowledge Catalog, BigQuery i Antigravity CLI do spłaszczania złożonych danych, programowego profilowania ich, generowania inteligentnych sugestii reguł jakości danych i wdrażania automatycznych skanów jakości. Głównym celem jest odejście od ręcznych procesów opartych na interfejsie, które są podatne na błędy i trudne do skalowania, oraz stworzenie solidnych, kontrolowanych pod względem wersji ram „zasad jako kodu”.
Czego się nauczysz
- Jak spłaszczyć zagnieżdżone dane BigQuery za pomocą widoków zmaterializowanych, aby umożliwić kompleksowe profilowanie.
- Jak programowo wywoływać skanowania profilujące Knowledge Catalog i nimi zarządzać za pomocą biblioteki klienta Knowledge Catalog w Pythonie.
- Jak eksportować dane profilu i strukturyzować je jako dane wejściowe dla modelu generatywnej AI.
- Jak przygotować prompt dla interfejsu Antigravity CLI, aby analizować dane profilu i generować plik reguł YAML zgodny z Knowledge Catalog.
- Znaczenie interaktywnego procesu z udziałem człowieka (HITL) w walidacji konfiguracji wygenerowanych przez AI.
- Jak wdrożyć wygenerowane reguły jako automatyczne skanowanie jakości danych.
Czego potrzebujesz
- Konto Google Cloud i projekt Google Cloud
- przeglądarka, np. Chrome;
Najważniejsze zagadnienia: filary jakości danych w Knowledge Catalog
Aby stworzyć skuteczną strategię jakości danych, musisz poznać podstawowe komponenty Knowledge Catalog.
- Skanowanie profilu danych: zadanie w Knowledge Catalog, które analizuje dane i generuje metadane statystyczne, w tym odsetek wartości null, liczbę różnych wartości i rozkłady wartości. Jest to nasza faza „odkrywania” w ramach automatyzacji.
- Reguły jakości danych: deklaratywne stwierdzenia określające warunki, które muszą spełniać Twoje dane (np.
NonNullExpectation,SetExpectation,RangeExpectation). - Generatywna AI do sugerowania reguł: wykorzystanie dużego modelu językowego (np. Gemini) do analizowania profilu danych i sugerowania odpowiednich reguł jakości danych. Przyspiesza to proces określania podstawowych ram jakości.
- Skanowanie jakości danych: zadanie Knowledge Catalog, które sprawdza poprawność danych pod kątem zestawu wstępnie zdefiniowanych lub niestandardowych reguł.
- Programowe zarządzanie: główny temat zarządzania kontrolami (np.regułami jakości) jako kodem (np. w plikach YAML i skryptach Pythona). Umożliwia to automatyzację, obsługa wersji i integrację z potokami CI/CD.
- Proces z udziałem człowieka: kluczowy punkt kontrolny, w którym wiedza i nadzór człowieka są włączane do zautomatyzowanego procesu. W przypadku konfiguracji wygenerowanych przez AI proces z udziałem człowieka jest niezbędny do sprawdzenia poprawności, przydatności biznesowej i bezpieczeństwa sugestii przed ich wdrożeniem.
2. Konfiguracja i wymagania
Uruchamianie Cloud Shell
Z Google Cloud można korzystać zdalnie na laptopie, ale w tym ćwiczeniu użyjesz Google Cloud Shell, czyli środowiska wiersza poleceń działającego w chmurze.
W konsoli Google Cloud kliknij ikonę Cloud Shell na pasku narzędzi w prawym górnym rogu:
Uzyskanie dostępu do środowiska i połączenie się z nim powinno zająć tylko kilka chwil. Po zakończeniu powinno wyświetlić się coś takiego:
Ta maszyna wirtualna zawiera wszystkie potrzebne narzędzia dla programistów. Zawiera również stały katalog domowy o pojemności 5 GB i działa w Google Cloud, co znacznie zwiększa wydajność sieci i usprawnia proces uwierzytelniania. Wszystkie zadania w tym laboratorium możesz wykonać w przeglądarce. Nie musisz niczego instalować.
Włączanie wymaganych interfejsów API i konfigurowanie środowiska
W Cloud Shell sprawdź, czy identyfikator projektu jest skonfigurowany:
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="kc_dq_codelab"
export TABLE_ID="ga4_transactions"
gcloud services enable dataplex.googleapis.com \
bigquery.googleapis.com \
serviceusage.googleapis.com \
aiplatform.googleapis.com
W tym przykładzie używamy lokalizacji us (wiele regionów), ponieważ publiczne dane przykładowe, których będziemy używać, również znajdują się w lokalizacji us (wiele regionów). BigQuery wymaga, aby dane źródłowe i tabela docelowa zapytania znajdowały się w tej samej lokalizacji.
Tworzenie dedykowanego zbioru danych BigQuery
Utwórz nowy zbiór danych BigQuery, w którym będą przechowywane przykładowe dane i wyniki.
bq --location=us mk --dataset $PROJECT_ID:$DATASET_ID
Przygotowywanie przykładowych danych
W tym samouczku użyjesz publicznego zbioru danych zawierającego maskowane dane e-commerce ze sklepu Google Merchandise Store. Publiczne zbiory danych są tylko do odczytu, więc musisz utworzyć w swoim zbiorze danych kopię, którą będzie można modyfikować. To polecenie bq tworzy nową tabelę ga4_transactions w zbiorze danych kc_dq_codelab. Kopiuje dane z jednego dnia (2021-01-31), aby skanowanie przebiegało szybko.
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`'
Konfigurowanie katalogu demonstracyjnego
Na początek sklonujesz repozytorium GitHub, które zawiera niezbędną strukturę folderów i pliki pomocnicze do tego ćwiczenia.
# Perform a shallow clone to get only the latest repository structure without the full history
git clone --depth 1 --filter=blob:none --sparse https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos
# Specify and download only the folder we need for this lab
git sparse-checkout set data-analytics/programmatic-dq
cd data-analytics/programmatic-dq
Ten katalog jest teraz Twoim aktywnym obszarem roboczym. Wszystkie kolejne pliki będą tworzone w tym miejscu.
3. Automatyczne wykrywanie danych za pomocą profilowania w Knowledge Catalog
Profilowanie danych w Knowledge Catalog to zaawansowane narzędzie do automatycznego wykrywania informacji statystycznych o danych, takich jak odsetek wartości null, unikalność i rozkłady wartości. Ten proces jest niezbędny do zrozumienia struktury i jakości danych. Znane ograniczenie profilowania w Knowledge Catalog polega jednak na tym, że nie można w pełni sprawdzić zagnieżdżonych ani powtarzających się pól (np. typów RECORD lub ARRAY) w tabeli. Może wykryć, że kolumna jest typem złożonym, ale nie może profilować poszczególnych pól w tej zagnieżdżonej strukturze.
Aby temu zapobiec, spłaszczymy dane w specjalnie utworzonych widokach zmaterializowanych. Dzięki tej strategii każde pole staje się kolumną najwyższego poziomu, co umożliwia usłudze Knowledge Catalog profilowanie każdego z nich osobno.
Informacje o zagnieżdżonym schemacie
Najpierw przyjrzyjmy się schematowi naszej tabeli źródłowej. Zbiór danych Google Analytics 4 (GA4) zawiera kilka zagnieżdżonych i powtarzających się kolumn. Aby programowo pobrać pełny schemat, w tym wszystkie struktury zagnieżdżone, możesz użyć polecenia bq show i zapisać dane wyjściowe jako plik JSON.
bq show --schema --format=json $PROJECT_ID:$DATASET_ID.$TABLE_ID > bq_schema.json
Sprawdzenie pliku bq_schema.json ujawnia złożone struktury, takie jak urządzenie, lokalizacja geograficzna, e-commerce i powtarzające się elementy rekordu. Są to struktury, które wymagają spłaszczenia, aby można było skutecznie utworzyć profil.
Spłaszczanie danych za pomocą widoków zmaterializowanych
Tworzenie zmaterializowanych widoków to najskuteczniejsze i najbardziej praktyczne rozwiązanie problemu zagnieżdżonych danych. Dzięki wstępnemu obliczaniu spłaszczonych wyników zmaterializowane widoki oferują znaczące korzyści w zakresie skuteczności i kosztów zapytań, a jednocześnie zapewniają analitykom i narzędziom do profilowania prostszą strukturę podobną do relacyjnej.
Pierwsza myśl może być taka, aby spłaszczyć wszystko w jeden gigantyczny widok. To intuicyjne podejście kryje jednak niebezpieczną pułapkę, która może prowadzić do poważnego uszkodzenia danych. Sprawdźmy, dlaczego jest to poważny błąd.
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`;
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;
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;
Teraz wykonaj te szablony za pomocą narzędzia wiersza poleceń bq. Polecenie envsubst odczyta każdy plik, zastąpi zmienne, takie jak $PROJECT_ID i $DATASET_ID, ich wartościami ze środowiska powłoki i przekieruje końcowy, prawidłowy kod SQL do polecenia bq query.
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
Wykonywanie skanów profilu za pomocą klienta Pythona
Teraz, gdy mamy już spłaszczone widoki, które można profilować, możemy programowo tworzyć i uruchamiać skanowania profilujące dane Knowledge Catalog dla każdego z nich. Poniższy skrypt w Pythonie używa biblioteki klienta google-cloud-dataplex do zautomatyzowania tego procesu.
Przed uruchomieniem skryptu warto utworzyć w katalogu projektu izolowane środowisko Pythona. Dzięki temu zależności projektu są zarządzane oddzielnie, co zapobiega konfliktom z innymi pakietami w środowisku powłoki Cloud Shell.
# Create the virtual environment
python3 -m venv dq_venv
# Activate the environment
source dq_venv/bin/activate
Teraz zainstaluj bibliotekę klienta Knowledge Catalog w nowo aktywowanym środowisku.
# Install the Knowledge Catalog client library
pip install google-cloud-dataplex
Po skonfigurowaniu środowiska i zainstalowaniu biblioteki możesz utworzyć skrypt orkiestracji.
Na pasku narzędzi Cloud Shell kliknij Otwórz edytor. Utwórz nowy plik o nazwie 1_run_scan.py i wklej do niego ten kod w Pythonie. Jeśli sklonujesz repozytorium GitHub, ten plik będzie już w Twoim folderze.
Ten skrypt utworzy skan dla każdego widoku zmaterializowanego (jeśli jeszcze nie istnieje), uruchomi go, a następnie będzie odpytywać, dopóki wszystkie zadania skanowania nie zostaną ukończone.
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()
Teraz uruchom skrypt z terminala Cloud Shell.
python 1_run_scan.py
Skrypt będzie teraz koordynować profilowanie 3 zmaterializowanych widoków, dostarczając aktualizacje stanu w czasie rzeczywistym. Po zakończeniu tego procesu uzyskasz bogaty, czytelny dla maszyn profil statystyczny każdego widoku, gotowy do następnego etapu naszego przepływu pracy: generowania reguł jakości danych opartych na AI.
Zakończone skanowanie profilu możesz wyświetlić w konsoli Google Cloud.
- W menu nawigacyjnym w sekcji Zarządzaj otwórz Knowledge Catalog i Profilowanie i jakość danych.
- Powinny się wyświetlić 3 skany profilu wraz z ich najnowszym stanem zadania. Aby wyświetlić szczegółowe wyniki, możesz kliknąć skan.
Od profilu BigQuery do danych wejściowych gotowych do AI
Skanowania profilujące w Knowledge Catalog zostały uruchomione. Wyniki są dostępne w interfejsie Knowledge Catalog API, ale aby użyć ich jako danych wejściowych dla modelu generatywnej AI, musimy wyodrębnić je do strukturalnego pliku lokalnego.
Ten skrypt w języku Python, 2_dq_profile_save.py, programowo wyszukuje najnowsze zakończone powodzeniem zadanie skanowania profilującego dla widoku mv_ga4_user_session_flat. Następnie pobiera pełny, szczegółowy wynik profilu i zapisuje go jako lokalny plik JSON o nazwie dq_profile_results.json. Ten plik będzie bezpośrednim wejściem do analizy AI w następnym kroku.
W edytorze Cloud Shell utwórz nowy plik o nazwie 2_dq_profile_save.py i wklej do niego ten kod: Podobnie jak w poprzednim kroku, możesz pominąć tworzenie pliku, jeśli sklonujesz repozytorium.
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 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 '1_run_scan.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()
Teraz uruchom skrypt w terminalu:
python 2_dq_profile_save.py
Po zakończeniu w katalogu pojawi się nowy plik o nazwie dq_profile_results.json. Ten plik zawiera bogate, szczegółowe metadane statystyczne, których użyjemy do wygenerowania reguł jakości danych. Jeśli chcesz sprawdzić zawartość pliku dq_profile_results.json, uruchom to polecenie:
cat dq_profile_results.json
4. Generowanie reguł jakości danych za pomocą Antigravity CLI
Teraz możesz użyć interfejsu wiersza poleceń Antigravity (agy), aby odczytać wyniki lokalnego skanowania profilu. Ręczne pisanie specyfikacji jakości danych w przypadku złożonych zbiorów danych jest czasochłonne i podatne na błędy. Użycie agenta generatywnej AI w terminalu przyspiesza ten proces, ponieważ w kilka sekund tworzy on wstępną konfigurację deklaratywną. Umożliwia to zespołom ds. danych przejście od ręcznego tworzenia składni do nadzoru z udziałem człowieka (HITL) na wysokim poziomie, zgodnego z potrzebami biznesowymi.
Aby uruchomić interfejs Antigravity CLI, uruchom w terminalu to polecenie:
agy
Możesz teraz wygenerować reguły jakości. Interfejs wiersza poleceń może odczytywać pliki w bieżącym katalogu, więc może bezpośrednio korzystać z nowych danych skanowania profilu.
Poproś agenta o utworzenie planu
Najpierw poprosimy agenta Antigravity o przeanalizowanie profilu statystycznego i zaproponowanie planu działania. Wyraźnie instruujemy, aby jeszcze nie zapisywał pliku YAML. Skupia się na analizie i uzasadnieniu.
W interaktywnej sesji Antigravity CLI wpisz ten prompt strukturalny:
# Context
You are preparing a data quality rule configuration plan for Google Cloud Knowledge Catalog based on data profile statistics.
# Input
- File Path: `./dq_profile_results.json` (contains metrics like null percentage, distinct counts, and distributions)
# Task
Analyze the input statistics and propose a step-by-step plan for establishing automated data quality rules.
*Do not write any YAML code in this step.* Focus only on analytical planning.
# Rule Mapping Strategy
For candidate columns, match the statistical metrics to the most appropriate expectations:
- `nonNullExpectation`: Propose for columns with 0% null values in the profile.
- `setExpectation`: Propose for columns with a highly limited, stable set of categorical values.
- `rangeExpectation`: Propose for numeric columns with consistent and predictable value boundaries.
# Guidelines
- Provide a metric-based justification for each proposed rule (e.g., "Recommend `nonNullExpectation` for column 'user_pseudo_id' because its null percentage is 0%").
- Flag volatile metrics such as hardcoded row counts that could cause false-positive alerts in production.
# Output Format
Provide your analysis and proposed rules as a structured, step-by-step markdown plan with clear headings.
Agent przeanalizuje plik JSON i powinien zwrócić uporządkowany plan podobny do tego:
Automated Data Quality Rule Configuration Plan
Google Cloud Knowledge Catalog (Dataplex Data Quality)
──────
## Executive Summary
This analytical planning document outlines a step-by-step strategy for configuring automated data quality (DQ) rules in Google Cloud Knowledge Catalog (formerly Dataplex Data Quality) based on profiling statistics.
The dataset contains 26,489 rows representing GA4 event logs. Based on statistical metrics (null ratios, distinct value distributions, and data types), candidate columns are mapped to appropriate expectation rules.
──────
## 1. Data Profile Overview & Statistical Highlights
Column Name │ Data Type │ Null Ratio │ Distinct Count │ Key Value Range / Categories
─────────────────┼───────────┼────────────────┼────────────────┼──────────────────────────────────────────────────
event_date │ STRING │ 0.0% (0) │ 1 (3.78e-05) │ "20210131" (100%)
event_timestamp │ INTEGER │ 0.0% (0) │ ~16,539 (0.62) │ Min: 1612051200657906, Max: 1612137595412363
event_name │ STRING │ 0.0% (0) │ 16 (0.0006) │ page_view (35.8%), user_engagement (18.9%), etc.
user_pseudo_id │ STRING │ 0.0% (0) │ ~2,545 (0.09) │ 18–21 characters string identifiers
user_id │ STRING │ 100.0% (1.0) │ 0 (0.0) │ Entirely NULL
device_category │ STRING │ 0.0% (0) │ 3 (0.0001) │ desktop (57.5%), mobile (40.1%), tablet (2.4%)
... │ ... │ ... │ ... │ ...
──────
## 2. Rule Mapping Strategy & Analytical Justifications
### Step 1: Nullability Rules (nonNullExpectation)
Propose nonNullExpectation for mandatory columns where the data profile demonstrates 0% null values.
• user_pseudo_id, event_timestamp, event_name, event_date, stream_id, platform, device_category (Metric Justification: nullRatio is 0.0%)
│ [!NOTE] Exclusions:
│ • user_id: Has a nullRatio of 100.0% (unauthenticated traffic).
│ • device_language: Has a nullRatio of 37.53%.
──────
### Step 2: Categorical Value Set Validation (setExpectation)
Propose setExpectation for columns with a highly limited, stable set of categorical domain values.
• device_category: Distinct count is exactly 3. Allowed set: ['desktop', 'mobile', 'tablet']
• platform: Distinct count is 1. Allowed set expanded to: ['WEB', 'ANDROID', 'IOS'] to avoid over-fitting.
• geo_continent: Distinct count is 6. Allowed set: ['Americas', 'Asia', 'Europe', 'Africa', 'Oceania', 'Antarctica', '(not set)']
──────
### Step 3: Numeric & Timestamp Boundary Validation (rangeExpectation)
Propose rangeExpectation for numeric columns with consistent and predictable value boundaries.
• event_timestamp: rangeExpectation requiring event_timestamp > 0 (avoid dynamic microsecond range hardcoding)
• stream_id: rangeExpectation requiring positive integer stream IDs (stream_id > 0)
──────
## 3. Risk Warning: Volatile Metrics & Production False Positives
│ [!WARNING] Volatile Metrics Flagged for Risk Mitigation:
1. Hardcoded Total Row Count (rowCount = 26,489) -> Daily event volume fluctuates. Use dynamic volume thresholds.
2. Hardcoded Partition Date (event_date = '20210131') -> Breaks on future runs. Validate against YYYYMMDD regex patterns.
3. Exact Timestamp Range Bounds -> Enforcing these microsecond limits on incoming live pipelines will reject all future data.
4. Single-Value Domain Restrictions -> Single profile sample might lack active streams. Set sets according to enterprise schema.
──────
## Summary Table of Proposed Rules
Target Column │ Rule Type │ Metric-Based Justification │ Operational Considerations
─────────────────┼────────────────────┼────────────────────────────┼──────────────────────────────────────────────────
user_pseudo_id │ nonNullExpectation │ Null Ratio: 0.0% │ Core identifier, strictly required
event_timestamp │ nonNullExpectation │ Null Ratio: 0.0% │ Temporal key, strictly required
event_timestamp │ rangeExpectation │ Min: > 0 (Microseconds) │ Avoid hardcoding epoch min/max
event_name │ nonNullExpectation │ Null Ratio: 0.0% │ Required event taxonomy key
event_name │ setExpectation │ Categorical distribution │ Map to standard GA4 event taxonomy
device_category │ nonNullExpectation │ Null Ratio: 0.0% │ Required form-factor dimension
device_category │ setExpectation │ Distinct Count: 3 values │ ['desktop', 'mobile', 'tablet']
... │ ... │ ... │ ...
Generowanie reguł jakości danych
Jest to najważniejszy krok w całym procesie: weryfikacja z udziałem człowieka. Plan wygenerowany przez agenta jest oparty wyłącznie na wzorcach statystycznych w danych. Nie ma ona wiedzy o kontekście Twojej firmy, przyszłych zmianach danych ani konkretnym celu, jaki przyświeca Ci przy ich używaniu. Twoim zadaniem jako eksperta jest sprawdzenie, poprawienie i zatwierdzenie tego planu przed przekształceniem go w kod.
Dokładnie sprawdź plan przedstawiony przez agenta.
- Czy to ma sens?
- Czy jest zgodna z Twoją wiedzą o firmie?
- Czy istnieją reguły, które są statystycznie poprawne, ale praktycznie bezużyteczne?
Wyniki uzyskane od agenta mogą się różnić. Twoim celem jest jego dopracowanie. Załóżmy na przykład, że plan sugeruje regułę rowCount, ponieważ tabela zawiera stałą liczbę wierszy w danych próbnych. Jako ekspert możesz wiedzieć, że rozmiar tej tabeli będzie codziennie rosnąć, co sprawia, że ścisła reguła dotycząca liczby wierszy jest niepraktyczna i może powodować fałszywe alerty. To doskonały przykład zastosowania kontekstu biznesowego, którego brakuje AI.
Teraz przekażesz agentowi opinię i wydasz mu ostateczne polecenie wygenerowania kodu. Musisz dostosować ten prompt do otrzymanego planu i wprowadzanych poprawek.
Prompt poniżej to szablon. W pierwszym wierszu podaj konkretne poprawki. Jeśli plan przedstawiony przez agenta jest idealny i nie wymaga zmian, możesz po prostu usunąć ten wiersz.
W tej samej sesji Antigravity wpisz dostosowaną wersję tego promptu:
# Feedback & Approvals
[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'."]
# Objective
Based on the approved analysis plan and the provided feedback, generate the final `dq_rules.yaml` file conforming to the standard `DataQualityRule` schema.
# Instructions
1. **Rule Justifications**: For every generated rule, add a YAML comment (`#`) on the line directly above it, briefly explaining the justification established in the plan.
2. **Schema Alignment**: Ensure the structure strictly adheres to the required Knowledge Catalog data quality scan specification. 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`) as the schema authority.
3. **Data-Driven Values**: Derive all rule parameters, such as thresholds or expected values, directly from the statistical metrics in `dq_profile_results.json`.
# Constraints
- **Output Purity**: Return ONLY the raw, valid, and properly formatted YAML code block.
- Do not include conversational preambles, introductory sentences, explanations, or markdown blocks around the YAML.
Agent wygeneruje teraz treść w formacie YAML na podstawie precyzyjnych, zweryfikowanych przez człowieka instrukcji. Po zakończeniu w katalogu roboczym pojawi się nowy plik o nazwie dq_rules.yaml.
Tworzenie i uruchamianie skanowania jakości danych
Teraz, gdy masz wygenerowany przez agenta i zweryfikowany przez człowieka plik dq_rules.yaml, możesz go bez obaw wdrożyć.
Aby zamknąć interfejs wiersza poleceń Antigravity, wpisz /quit lub dwukrotnie naciśnij Ctrl+C.
To polecenie gcloud tworzy nowy zasób skanowania danych Knowledge Catalog. Nie uruchamia jeszcze skanowania, a jedynie rejestruje definicję i konfigurację skanowania (nasz plik YAML) w Knowledge Catalog.
Uruchom to polecenie w terminalu:
export DQ_SCAN="dq-scan"
gcloud dataplex datascans create data-quality $DQ_SCAN \
--project=$PROJECT_ID \
--location=$LOCATION \
--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"
Po zdefiniowaniu skanowania możesz uruchomić zadanie, aby je wykonać.
gcloud dataplex datascans run $DQ_SCAN --location=$LOCATION --project=$PROJECT_ID
To polecenie zwróci identyfikator zadania. Stan tego zadania możesz monitorować w sekcji Knowledge Catalog w konsoli Google Cloud. Po zakończeniu wyniki zostaną zapisane w tabeli BigQuery do analizy.
5. Kluczowa rola procesu z udziałem człowieka
Korzystanie z agenta Antigravity do przyspieszania generowania reguł jest niezwykle przydatne, ale ważne jest, aby traktować AI jako wykwalifikowanego asystenta, a nie w pełni autonomicznego pilota. Proces z udziałem człowieka nie jest opcjonalną sugestią, ale niepodlegającym negocjacjom podstawowym krokiem w każdym przepływie pracy związanym z zarządzaniem danymi. Wdrażanie artefaktów wygenerowanych przez AI bez rygorystycznego nadzoru człowieka to przepis na porażkę.
Traktuj wygenerowane przez AI dq_rules.yaml jako żądanie pull przesłane przez niezwykle szybkiego, ale niedoświadczonego programistę AI. Zanim będzie można go scalić z „główną gałęzią” zasad zarządzania i wdrożyć, musi zostać dokładnie sprawdzony przez starszego eksperta – Ciebie. Jest to niezbędne do ograniczenia słabych stron dużych modeli językowych.
Oto szczegółowe wyjaśnienie, dlaczego weryfikacja manualna jest niezbędna i na co należy zwrócić szczególną uwagę:
1. Walidacja kontekstowa: AI nie zna kontekstu biznesowego
- Słaba strona LLM: LLM to mistrz wzorców i statystyk, ale nie rozumie kontekstu Twojej firmy. Jeśli na przykład kolumna
new_campaign_idma współczynnik wartości null wynoszący 98%, LLM może ją zignorować ze względów statystycznych. - Kluczowa rola człowieka: Ty, jako ekspert, wiesz, że pole
new_campaign_idzostało dodane dopiero wczoraj w związku z ważnym wprowadzeniem produktu na rynek w przyszłym tygodniu. Wiesz, że współczynnik wartości null powinien być teraz wysoki, ale oczekuje się, że znacznie spadnie. Wiesz też, że po wypełnieniu musi mieć określony format. AI nie jest w stanie wywnioskować tej zewnętrznej wiedzy biznesowej. Twoim zadaniem jest zastosowanie tego kontekstu biznesowego do sugestii statystycznych AI, zastępując je lub uzupełniając w razie potrzeby.
2. Poprawność i precyzja: ochrona przed halucynacjami i subtelnymi błędami
- Słaba strona LLM: modele LLM mogą być „pewne, że mają rację, ale się mylić”. Mogą one „halucynować” lub generować kod, który jest subtelnie nieprawidłowy. Może na przykład wygenerować plik YAML z prawidłowo nazwaną regułą, ale nieprawidłowym parametrem, lub może błędnie napisać typ reguły (np.
setExpectationszamiast prawidłowegosetExpectation). Te subtelne błędy spowodują niepowodzenie wdrożenia, ale mogą być trudne do wykrycia. - Kluczowa rola człowieka: Twoim zadaniem jest pełnienie funkcji ostatecznego narzędzia do sprawdzania kodu i walidacji schematu. Musisz dokładnie sprawdzić wygenerowany plik YAML pod kątem zgodności z oficjalną
DataQualityRulespecyfikacją Knowledge Catalog. Nie sprawdzasz tylko, czy „wygląda dobrze”, ale weryfikujesz poprawność składniową i semantyczną, aby mieć pewność, że jest w 100% zgodna z docelowym interfejsem API. Dlatego w tym samouczku agent jest proszony o odwoływanie się do plików schematu, aby zmniejszyć ryzyko błędu. Ostateczna weryfikacja należy jednak do Ciebie.
3. Bezpieczeństwo i ograniczanie ryzyka: zapobieganie konsekwencjom
- Słaba strona LLM: wdrożenie w środowisku produkcyjnym wadliwej reguły dotyczącej jakości danych może mieć poważne konsekwencje. Jeśli AI zaproponuje
rangeExpectationdla kwoty transakcji finansowej, która jest zbyt szeroka, może nie wykryć nieuczciwego działania. Z drugiej strony, jeśli na podstawie małej próbki danych zasugeruje zbyt rygorystyczną regułę, może zasypać zespół dyżurny tysiącami fałszywie pozytywnych alertów, co doprowadzi do zmęczenia alertami i spowoduje przeoczenie prawdziwych problemów. - Kluczowa rola człowieka: to Ty jesteś inżynierem ds. bezpieczeństwa. Musisz ocenić potencjalny wpływ każdej reguły sugerowanej przez AI. Zastanów się: „Co się stanie, jeśli ta reguła nie zadziała? Czy alert wymaga podjęcia działań? Jakie jest ryzyko, jeśli ta reguła zostanie nieprawidłowo zastosowana? Ocena ryzyka to unikalna umiejętność człowieka, która polega na porównaniu kosztów niepowodzenia z korzyściami wynikającymi z weryfikacji.
4. Zarządzanie jako proces ciągły: uwzględnianie wiedzy wybiegającej w przyszłość
- Słaba strona LLM: wiedza AI opiera się na statycznym zrzucie danych – profil jest wynikiem z określonego momentu. Nie ma informacji o przyszłych wydarzeniach.
- Kluczowa rola człowieka: Twoja strategia nadzoru musi być przyszłościowa. Wiesz, że w przyszłym miesiącu źródło danych zostanie przeniesione, co spowoduje zmianę identyfikatora stream_id. Wiesz, że do listy
geo_countrydodawany jest nowy kraj. Proces HITL polega na wprowadzaniu tej wiedzy o przyszłości, aktualizowaniu lub tymczasowym wyłączaniu reguł, aby zapobiec przerwom w działaniu podczas planowanych zmian w firmie lub technologii. Jakość danych nie jest jednorazową konfiguracją. To proces, który musi się rozwijać, a tylko człowiek może nim kierować.
Podsumowując, HITL to niezbędny mechanizm zapewniania jakości i bezpieczeństwa, który przekształca zarządzanie oparte na AI z nowatorskiego, ale ryzykownego pomysłu w odpowiedzialną, skalowalną i klasy korporacyjnej praktykę. Dzięki temu wdrożone zasady są nie tylko przyspieszone przez AI, ale także zweryfikowane przez człowieka, co łączy szybkość maszyn z wiedzą i kontekstem ekspertów.
Nacisk na nadzór człowieka nie umniejsza jednak wartości AI. Wręcz przeciwnie, generatywna AI odgrywa kluczową rolę w przyspieszaniu samego procesu HITL.
Bez AI inżynier danych musiałby:
- Ręczne pisanie złożonych zapytań SQL w celu profilowania danych (np.
COUNT DISTINCT,AVG,MIN,MAXdla każdej kolumny). - dokładnie analizować wyniki w poszczególnych arkuszach kalkulacyjnych,
- Każdy wiersz pliku reguł YAML trzeba napisać od zera, co jest żmudne i podatne na błędy.
Sztuczna inteligencja automatyzuje te żmudne i czasochłonne czynności. Działa on jak niestrudzony analityk, który natychmiast przetwarza profil statystyczny i tworzy dobrze zorganizowany, w 80% gotowy „pierwszy szkic” zasad.
Zasadniczo zmienia to charakter pracy człowieka. Zamiast spędzać godziny na ręcznym przetwarzaniu danych i pisaniu powtarzalnego kodu, ekspert może od razu skupić się na zadaniach o największej wartości:
- stosowanie krytycznego kontekstu biznesowego,
- Sprawdzanie poprawności logiki AI.
- podejmowanie strategicznych decyzji dotyczących tego, które reguły są naprawdę ważne;
W ramach tej współpracy AI zajmuje się kwestią „co” (jakie są wzorce statystyczne), dzięki czemu człowiek może skupić się na pytaniach „dlaczego” (dlaczego ten wzorzec jest ważny dla naszej firmy?) i „co z tego wynika” (jakie powinny być nasze zasady?). Dlatego AI nie zastępuje pętli, tylko sprawia, że każdy cykl jest szybszy, bardziej inteligentny i skuteczniejszy.
6. Czyszczenie środowiska
Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby zużyte w tym ćwiczeniu, usuń projekt zawierający te zasoby. Jeśli jednak chcesz zachować projekt, możesz usunąć utworzone zasoby.
Usuwanie skanów Knowledge Catalog
Najpierw usuń utworzone przez siebie skanowania profilu i jakości. Aby zapobiec przypadkowemu usunięciu ważnych zasobów, w tych poleceniach używane są konkretne nazwy skanów utworzonych w tym laboratorium.
# 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 datascans delete profile-scan-mv-ga4-user-session-flat \
--location=us-central1 \
--project=$PROJECT_ID --quiet
gcloud dataplex datascans delete profile-scan-mv-ga4-ecommerce-transactions \
--location=us-central1 \
--project=$PROJECT_ID --quiet
gcloud dataplex datascans delete profile-scan-mv-ga4-ecommerce-items \
--location=us-central1 \
--project=$PROJECT_ID --quiet
Usuwanie zbioru danych BigQuery
Następnie usuń zbiór danych BigQuery. To polecenie jest nieodwracalne i używa flagi -f (force), aby usunąć zbiór danych i wszystkie jego tabele bez potwierdzenia.
# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:kc_dq_codelab
7. Gratulacje!
Udało Ci się ukończyć ćwiczenie w Codelabs.
Udało Ci się utworzyć kompleksowy, programowy przepływ pracy związany z zarządzaniem danymi. Na początku używasz widoków zmaterializowanych do spłaszczania złożonych danych BigQuery, aby nadawały się do analizy. Następnie programowo przeprowadzono skanowania profilujące Knowledge Catalog, aby wygenerować metadane statystyczne. Co najważniejsze, wykorzystano interfejs Antigravity CLI do analizowania danych wyjściowych profilu i inteligentnego generowania artefaktu „zasady jako kod” (dq_rules.yaml). Następnie za pomocą interfejsu CLI wdrożono tę konfigurację jako automatyczne skanowanie jakości danych, co pozwoliło zamknąć pętlę nowoczesnej, skalowalnej strategii zarządzania.
Znasz już podstawowy wzorzec tworzenia w Google Cloud niezawodnych systemów jakości danych, które są przyspieszane przez AI i weryfikowane przez ludzi.
Co dalej?
- Integracja z CI/CD: weź plik
dq_rules.yamli zatwierdź go w repozytorium Git. Utwórz potok CI/CD (np. za pomocą Cloud Build lub GitHub Actions), który automatycznie wdraża skanowanie Knowledge Catalog za każdym razem, gdy plik reguł zostanie zaktualizowany. - Poznaj niestandardowe reguły SQL: wyjdź poza standardowe typy reguł. Knowledge Catalog obsługuje niestandardowe reguły SQL, które umożliwiają wymuszanie bardziej złożonej logiki specyficznej dla firmy, której nie można wyrazić za pomocą predefiniowanych kontroli. Ta funkcja umożliwia dostosowanie weryfikacji do Twoich unikalnych wymagań.
- Optymalizuj skanowanie pod kątem wydajności i kosztów: w przypadku bardzo dużych tabel możesz zwiększyć wydajność i obniżyć koszty, nie skanując zawsze całego zbioru danych. Wypróbuj filtry, aby zawęzić skanowanie do określonych przedziałów czasowych lub segmentów danych, albo skonfiguruj skanowanie próbkowane, aby sprawdzić reprezentatywny odsetek danych.
- Wizualizacja wyników: dane wyjściowe każdego skanowania jakości danych w Knowledge Catalog są zapisywane w tabeli BigQuery. Połącz tę tabelę z Looker Studio, aby tworzyć panele, które śledzą wyniki jakości danych w czasie, zagregowane według zdefiniowanych przez Ciebie wymiarów (np. kompletność, ważność). Dzięki temu monitorowanie jest proaktywne i widoczne dla wszystkich zainteresowanych stron.
- Dziel się sprawdzonymi metodami: zachęcaj do wymiany wiedzy w organizacji, aby wykorzystywać zbiorowe doświadczenie i ulepszać strategię jakości danych. Budowanie kultury zaufania do danych jest kluczem do maksymalnego wykorzystania działań związanych z zarządzaniem.
- Przeczytaj dokumentację: