Collecte et retraitement

Skill de référence : .claude/skills/lakehouse-conventions/SKILL.md. Code : ingest/, dbt/, data/.

Couches

Couche	Contenu	Colonnes obligatoires	Stockage
landing	Fichiers bruts immuables, `landing/<source>/<date>/`	(fichiers)	MinIO, bucket `snowlab`
raw	Une table Iceberg par source et par type de fichier	`_source_system`, `_source_file`, `_row_number`, `_loaded_at` (plus `_row_hash`)	Iceberg, schéma `raw`
standardized	Modèle canonique long (entité, métrique, période)	colonnes de preuve plus `period_start`, `period_end`, `currency`, `unit`, `confidence_class`	dbt, tables Iceberg dans Polaris (`analytics.standardized.*`)
semantic	Référentiel des établissements, comp sets, faits quotidiens, KPI MetricFlow	selon le modèle	dbt, tables Iceberg dans Polaris (`analytics.semantic.*`)
intake	Tables techniques : historique de chargement, registre de schémas, exceptions, file de revue, suggestions de mapping, règles de mapping	selon la table	Iceberg (schéma `intake`) ; seed `mapping_rules` écrit par dbt dans le même schéma

Points d'entrée

Opération	Point d'entrée
Générer les fichiers d'une date	CLI `python -m data.generate landing --root <dossier> --as-of <date> [--history-days N]`
Déposer en landing	API Python `LandingStore.put` (pas de CLI)
Détecter, classer et charger en raw	CLI `python -m ingest.raw_loader --date <date> [--source <source>] [--raw-schema <schéma>]` (défaut `raw`) ; API `process_landing_date`, `RawLoader.load`
Retraiter	`dbt build` (ou `dbt seed`, `dbt snapshot`) sur `dbt/`
Suggestions de mapping	API `MappingSuggestions.suggest` (pas de CLI)
Exceptions des tests dbt	API `IntakeLog.record_dbt_failures(run_results_path)` (pas de CLI)
Décision humaine	SQL `UPDATE analytics.intake.review_queue ...` dans Trino, ou API `IntakeLog.set_review_status`
Snapshot `shared_calendar`	API `write_calendar_snapshot` (pas de CLI)
Lineage des chargements	CLI `ingest.raw_loader` (émission dès que `MARQUEZ_URL` est défini ; sinon une ligne INFO « lineage disabled, MARQUEZ_URL not set » et chargement sans lineage) ; API `LineageEmitter.from_env(bucket, database, raw_schema)` passé à `RawLoader(lineage=...)` (écart corrigé le 2026-10-23 : la CLI ne branchait pas l'émetteur)

Landing

ingest/landing.py, classe LandingStore (enveloppe d'un système de fichiers pyarrow : S3FileSystem sur MinIO via LandingStore.for_minio(cfg), LocalFileSystem dans les tests via LandingStore.local(dir)).

Clé : landing/<source>/<date ISO>/<nom>. Source conforme à ^[a-z][a-z0-9_]{0,63}$, nom de fichier à ^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$.
put refuse d'écraser un fichier existant (ImmutableFileError) : la couche landing est immuable.
list(source, date) liste les fichiers d'une date (fichiers cachés ignorés) et calcule pour chacun SHA-256 et taille (LandedFile).
LandedFile.key est la valeur de _source_file ; LandedFile.location (s3://snowlab/...) est le chemin lu par DuckDB.

Équivalent Snowflake : stage interne et PUT (https://docs.snowflake.com/en/user-guide/data-load-local-file-system-create-stage, vérifié le 2026-10-22).

Détection de schéma

ingest/detect.py :

CSV : sniff_csv(?) de DuckDB (délimiteur, en-tête, colonnes et types).
JSON, JSONL, NDJSON : DESCRIBE SELECT * FROM read_json_auto(?).
Autre extension : ValueError.
normalize_name transforme un nom de colonne source en identifiant minuscule non quoté (symboles remplacés par _, préfixe c_ si le nom commence par un chiffre).
fingerprint(columns) : SHA-256 de la liste ordonnée des paires (nom, type). Tout renommage, changement de type, ajout, suppression ou réordonnancement change l'empreinte.

Classification des fichiers

ingest/classify.py. Trois types connus (FILE_TYPES) :

Type	Source	Table raw	Motif de nom	Colonnes requises	Mesures rapprochées
`pms_daily`	`pms`	`pms_daily`	`^pms_.*\.csv$`	`hotel_id`, `business_date`, `rooms_available`, `rooms_sold`, `room_revenue`	`rooms_available`, `rooms_sold`, `room_revenue`
`bookings_reservations`	`bookings`	`bookings_reservations`	`^bookings_.*\.(json\\|jsonl\\|ndjson)$`	`hotel_id`, `reservation_id`, `stay_date`, `rooms`, `room_revenue`	`rooms`, `room_revenue`
`rate_shop_rates`	`rate_shop`	`rate_shop_rates`	`^rate_shop_.*\.csv$`	`hotel_id`, `competitor_id`, `stay_date`, `rate`, `shopped_at`	`rate`

Algorithme de classify(file_name, schema, client=None, *, use_ai=True) :

Règles : si exactement un type correspond à la fois par le nom et par les colonnes, il est retenu (méthode rules).
Sinon, si use_ai : appel d'AI_CLASSIFY (cœur cortex_lab) avec une description textuelle du fichier (nom et colonnes) et les trois types en catégories (label et description de 25 mots maximum), plus une task_description.
Le type proposé par le modèle n'est retenu que si le fichier porte réellement ses colonnes requises (méthode ai). Le modèle propose, les règles disposent.
Sinon : non classé (méthode none), ce qui écrit une exception unclassified_file et un élément de file de revue.

Registre de schémas

ingest/schema_registry.py, table analytics.intake.schema_registry (ajout seul) : registry_id, source, file_type, file_date, fingerprint, columns_json, change_type (new, unchanged, changed), diff_json (colonnes ajoutées, retirées, retypées, réordonnancement), detected_at.

SchemaRegistry.register compare au dernier schéma enregistré de la même source et du même type ; une observation identique à la même date n'est pas réécrite.
Un changement écrit une exception schema_change.
evolve_iceberg_schema(table, incoming) : ajoute les colonnes nouvelles (optionnelles), promeut int en long et float en double, laisse les colonnes disparues dans la table (NULL pour les nouvelles lignes). Tout autre changement de type lève SchemaConflict : le fichier est rejeté (rejected) avec une exception schema_conflict de sévérité error.

Chargement raw (rôle de `COPY INTO` et Snowpipe)

ingest/raw_loader.py, classe RawLoader(catalog, con, *, clock, lineage=None, memory_days=14, raw_schema="raw").

Colonnes de preuve et clé de hash

Définies dans ingest/__init__.py (EVIDENCE_COLUMNS, HASH_COLUMN) :

Colonne	Contenu	Équivalent Snowflake
`_source_system`	Nom de la source (`pms`, `bookings`, `rate_shop`)	aucun
`_source_file`	Clé de l'objet déposé, relative au bucket (`landing/pms/2026-10-22/pms_h001.csv`)	`METADATA$FILENAME`
`_row_number`	Numéro de ligne de données, à partir de 1, en-tête exclu	`METADATA$FILE_ROW_NUMBER`
`_loaded_at`	Horodatage UTC du chargement (microsecondes)	aucun
`_row_hash`	SHA-256 des valeurs des colonnes de données de la ligne	aucun (clé de hash pour les doublons)

Référence : https://docs.snowflake.com/en/user-guide/querying-metadata (vérifié le 2026-10-22).

Déroulé de `load(landed, file_type, schema=None)`

Mémoire de chargement : si intake.load_history contient un chargement réussi du même chemin depuis moins de 14 jours, le fichier est ignoré (skipped_already_loaded), comme Snowpipe. Si le contenu a changé depuis (SHA-256 différent), une exception file_modified_after_load est écrite, et le fichier n'est pas rechargé. Référence : https://docs.snowflake.com/en/user-guide/data-load-snowpipe-intro (vérifié le 2026-10-22).
Événement de lineage START (si un émetteur est fourni).
Détection du schéma, lecture du fichier par DuckDB (read_csv ou read_json_auto), noms normalisés, valeurs imbriquées converties en JSON compact (pas de variant en Iceberg v2), timestamps ramenés en microsecondes (UTC pour ceux qui portent un fuseau).
Enregistrement dans le registre de schémas (exception schema_change si changement).
Ajout des colonnes de preuve et du hash, création de la table raw si absente, évolution du schéma Iceberg (rejet sur conflit).
Détection des doublons par _row_hash : exception duplicate_rows (les lignes sont tout de même chargées ; standardized les dédoublonne).
Écriture par PyIceberg, colonnes alignées sur le schéma de la table (colonne absente du fichier : NULL).
Rapprochement des totaux : somme de chaque mesure dans le fichier contre la somme chargée en raw pour ce fichier ; écart supérieur à 1e-6 : exception total_mismatch de sévérité error.
Ligne dans intake.load_history (file_key, file_sha256, source, raw_table, loaded_at, row_count, status), puis événement COMPLETE.

process_landing_date(loader, store, source, date, *, client=None, use_ai=True) détecte, classe et charge tous les fichiers d'une date.

Exceptions et file de revue

ingest/exceptions.py, classe IntakeLog :

analytics.intake.exceptions : exception_id, detected_at, source, file_key, rule, severity (warning ou error), detail (JSON).
analytics.intake.review_queue : item_id, item_type (exception ou mapping_suggestion), ref_id, created_at, status (pending, accepted, rejected, resolved), reviewer, decided_at, comment.

Règles : unclassified_file, total_mismatch, duplicate_rows, schema_change, file_modified_after_load, schema_conflict, anomaly, confidence_class_violation, ainsi que mapping_suggestion_failed (écrite par ingest/mapping.py).

record_dbt_failures(run_results_path) lit target/run_results.json et écrit une exception par test dbt en échec ou en erreur, avec une règle déduite du nom du test (reconcile vers total_mismatch, duplicate et unique vers duplicate_rows, confidence_class et missing_value vers confidence_class_violation, sinon anomaly).

Décision humaine (mode retenu : mise à jour par SQL du statut) :

UPDATE analytics.intake.review_queue
SET status = 'accepted', reviewer = '<nom>', decided_at = current_timestamp(6)
WHERE item_id = '<id>';

dans Trino avec l'utilisateur lab_tech, ou IntakeLog.set_review_status(item_id, status, reviewer, comment=None) (upsert PyIceberg).

Mapping des champs

ingest/mapping.py.

Règles versionnées : seed dbt/seeds/mapping_rules.csv (rule_version, source, file_type, source_column, target_field, transform). Seule la transformation identity existe. Toute cible doit être un champ canonique : entity_id, competitor_id, period_start, currency, rooms_available, rooms_sold, room_revenue, reservation_id, reservation_rooms, reservation_revenue, booked_on, status, as_of_ts, competitor_rate.
load_rules, rules_for (dernière version par défaut), apply_mapping (renomme, garde preuves et hash, NULL pour une colonne source absente, écarte les colonnes sans règle), unmapped_columns.
Suggestions : MappingSuggestions.suggest(source, file_type, samples_by_column, rule_version) fait un appel AI_COMPLETE par colonne sans règle, avec le modèle de CORTEX_COMPLETE_MODEL (défaut cortex-complete), temperature 0, max_tokens 300 et une sortie structurée (target_field dans l'énumération des champs canoniques ou none, confidence entre 0 et 1, reason). Chaque suggestion est écrite dans analytics.intake.mapping_suggestions et mise en file de revue au statut pending. Une réponse hors contrat ou une panne de passerelle écrit une exception mapping_suggestion_failed, sans suggestion.
Jamais appliquée automatiquement : accepted_seed_rows() renvoie les lignes à ajouter au seed (avec rule_version + 1) pour les suggestions acceptées dans la file ; leur report dans mapping_rules.csv se fait par un commit.

Les modèles dbt standardized écrivent eux-mêmes leurs renommages en SQL ; la cohérence entre le seed et ces modèles n'est pas contrôlée automatiquement. Savoir si elle doit l'être est une question de conception ouverte (point V21, décision A11 de l'avenant du 2026-10-23, non traitée).

Projet dbt

Un seul projet, dbt/ (nom lab, profil lab), pour trois adapters.

Profils (`dbt/profiles.example.yml`)

Cible	Adapter	Rôle
`duckdb` (défaut)	`dbt-duckdb`	Base DuckDB locale `:memory:`, Polaris attaché sous l'alias `LAB_DATABASE` (`analytics`, aussi valeur de `database`) avec `access_delegation_mode: none` ; dbt écrit donc dans Polaris. Extensions `httpfs` et `iceberg`, plugin `cortex_lab.dbt_plugin`, secrets S3 et Iceberg lus dans l'environnement. `disable_transactions: true` : Polaris 1.8.0 refuse le commit multi-tables que DuckDB envoie pour une transaction dbt
`trino`	`dbt-trino`	Trino sans authentification (`method: none`), utilisateur `DBT_TRINO_USER`, base `LAB_DATABASE`
`snowflake`	`dbt-snowflake`	Côté client, identifiants `SNOWFLAKE_*` lus dans l'environnement

profiles.yml est ignoré par git ; l'exemple ne contient aucun secret.

Configuration (`dbt/dbt_project.yml`)

quoting désactivé pour la base, le schéma et l'identifiant.
Variables : raw_database (vide : base de la cible, donc analytics.raw.* pour les trois adapters), raw_schema (raw ; les tests d'intégration chargent dans un schéma scratch) et schema_prefix (vide ; les tests d'intégration le fixent pour écrire dans des namespaces préfixés au lieu des vraies couches).
models/standardized et models/semantic matérialisés en tables dans les schémas du même nom ; seeds hotels et comp_sets dans semantic, mapping_rules dans intake ; snapshots dans standardized (clé +schema, pas target_schema : seule schema passe par generate_schema_name, donc par le préfixe).
macros/generate_schema_name.sql : le schéma personnalisé est utilisé tel quel (pas de préfixe de cible), pour garder les noms des schémas Snowflake, précédé de schema_prefix quand il est défini.

Macros

Macro	Rôle
`get(expr, field)` (`macros/get.sql`)	Dispatch : `get(...)` natif sur Snowflake et via `cortex_lab` sur DuckDB, `json_extract_scalar(expr, '$.field')` sur Trino
`confidence_classes()`, `confidence_observed(expr)`, `confidence_calculated(expr)` (`macros/confidence_class.sql`)	Domaine fermé de `confidence_class` ; valeur présente : `observed` ou `calculated`, valeur absente : NULL avec `unknown`
`duckdb__drop_relation` (`macros/drop_relation.sql`)	Repli pour Polaris : suppression sans `CASCADE` (DuckDB le refuse sur une table Iceberg). Préfixée `duckdb__`, sans effet sur `dbt-trino` et `dbt-snowflake`
`duckdb__make_intermediate_relation` (`macros/intermediate_relation.sql`)	Correctif pour Polaris : nom intermédiaire `<modèle>__dbt_tmp_<début de invocation_id>`. Un renommage ne déplace pas les fichiers, la table garde l'emplacement de son intermédiaire, et Polaris refuserait (403) de recréer `__dbt_tmp` à cet emplacement à l'exécution suivante. Les réglages de chevauchement d'emplacements de Polaris ne sont pas modifiés
tests génériques `confidence_class_domain`, `missing_value_is_unknown`, `unique_combination`	Domaine fermé, cohérence NULL et `unknown`, unicité d'une combinaison de colonnes

Couche standardized

Format long : une ligne par entité, métrique et période, avec les colonnes de preuve, _row_hash, entity_id, metric, value (decimal(38, 4)), period_start, period_end (lendemain, via dbt.dateadd), currency (seulement pour les montants), unit (rooms, money, ratio), confidence_class, as_of_ts.

Modèle	Source raw	Métriques	Règles
`std_pms_daily`	`pms_daily`	`rooms_available`, `rooms_sold`, `room_revenue` (`observed`), `occupancy` (`calculated` : `rooms_sold / rooms_available`)	Le dernier export d'une date d'activité gagne (tri sur `_loaded_at`, `_source_file`, `_row_number`) : corrections tardives
`std_bookings`	`bookings_reservations`	`reservation_rooms`, `reservation_revenue`	Doublons intra-fichier (même hash) comptés une fois ; devise reprise du référentiel `hotels` ; `as_of_ts` = `extracted_at`
`std_rate_shop`	`rate_shop_rates`	`competitor_rate`	Doublons intra-fichier comptés une fois ; `as_of_ts` = `shopped_at`

confidence_class prend exactement une des six valeurs observed, calculated, estimated, proxy, inferred, unknown. Une donnée manquante reste NULL avec unknown, jamais une valeur inventée.

Historique long : SCD2

Snapshots dbt, stratégie timestamp sur as_of_ts, clé snapshot_key = entity_id || '-' || period_start :

snap_bookings_otb : réservations confirmées à bord (rooms_on_the_books, revenue_on_the_books) par hôtel et date de séjour, à la dernière extraction de chaque hôtel.
snap_rate_shop : tarifs du comp set (comp_rate_avg, comp_rate_min, comp_rate_max, comp_rate_count) par hôtel et date de séjour, au dernier relevé.

Chaque version est valable de dbt_valid_from à dbt_valid_to : la valeur connue à un moment passé se relit, ce que la prévision utilise pour ses covariables as-of (voir prevision.md).

Couche semantic

Modèle	Contenu
`dim_hotel`	Référentiel des établissements (seed `hotels`)
`dim_comp_set`	Concurrents suivis par établissement (seeds `comp_sets` et `hotels`)
`fct_hotel_daily`	Une ligne par hôtel et date d'activité : `rooms_available`, `rooms_sold`, `room_revenue`, `occupancy`, `adr`, `revpar` (NULL si dénominateur nul ou absent)
`metricflow_time_spine`	Calendrier quotidien 2020-01-01 à 2030-12-31 en SQL pur

Modèle sémantique MetricFlow (models/semantic/semantic_models.yml) : modèles hotel_daily (mesures rooms_available, rooms_sold, room_revenue, dimension temporelle business_date) et hotel (dimensions city, hotel_name) ; métriques rooms_sold_total, room_revenue_total, rooms_available_total, occupancy_rate, adr, revpar. Équivalent Snowflake : semantic views (https://docs.snowflake.com/en/user-guide/views-semantic/overview et https://docs.snowflake.com/en/sql-reference/sql/create-semantic-view, vérifiés le 2026-10-22) ; la traduction en CREATE SEMANTIC VIEW n'est pas faite dans le dépôt.

Tests de qualité

Outil retenu : tests dbt seulement (pas de Soda Core).

Test	Contrôle
`not_null`, `confidence_class_domain`, `missing_value_is_unknown` sur chaque modèle standardized	Colonnes de preuve et clés présentes, domaine fermé, NULL avec `unknown`
`unique_combination`	Unicité des clés de chaque modèle standardized et semantic
`relationships` sur `fct_hotel_daily.hotel_id`	Intégrité vers `dim_hotel`
`tests/assert_pms_totals_reconcile.sql`	Total de chiffre d'affaires raw (dernier export) égal au total standardized
`tests/assert_no_duplicate_pms_keys.sql`	Une valeur par hôtel, date et métrique
`tests/assert_occupancy_in_bounds.sql`	Taux d'occupation entre 0 et 1
`tests/assert_no_negative_amounts.sql`	Aucun montant, nombre de chambres ou tarif négatif

Les échecs peuvent alimenter la table d'exceptions via IntakeLog.record_dbt_failures.

Lineage

ingest/lineage.py, classe LineageEmitter (client openlineage-python) :

Un chargement raw est un job ingest.raw_load.<table raw> dans l'espace de noms snowlab : START, puis COMPLETE (avec la statistique rowCount) ou FAIL.
Entrée : dataset s3://snowlab / clé du fichier ; sortie : dataset iceberg://analytics / <schéma raw>.<table> (raw par défaut, le schéma réel du chargement avec --raw-schema).
LineageEmitter.from_env(bucket, database, raw_schema) renvoie un émetteur si MARQUEZ_URL est défini, sinon None. La CLI l'appelle : sans MARQUEZ_URL, une ligne INFO l'indique et le chargement se fait sans lineage, sans erreur ni autre code de sortie.
Ni l'URL de Marquez avec ses identifiants éventuels ni un en-tête d'authentification ne sont journalisés.
Une erreur d'émission est journalisée et ignorée : le lineage ne casse jamais un chargement.
Les exécutions dbt se tracent avec dbt-ol (openlineage-dbt==1.54.0), vérifié le 2026-10-23 (B4) : dbt-ol build --target duckdb avec OPENLINEAGE_URL et OPENLINEAGE_NAMESPACE envoie les jobs et les datasets des modèles à Marquez. Écart constaté : les datasets de dbt-ol portent l'espace de noms duckdb://:memory: (nom de la connexion DuckDB), alors que le chargeur émet iceberg://analytics ; Marquez ne relie donc pas les datasets raw lus par dbt à ceux écrits par le chargeur. Aucun changement du code d'émission dans l'avenant ; l'écart est documenté et testé (tests/ingest/test_dbt_lineage_integration.py).

Marquez (marquezproject/marquez:0.51.1) tourne en API seule (port 5000, pas d'OpenSearch, pas d'interface web), avec migration de schéma au démarrage dans snowlab_marquez_db. Lecture : GET /api/v1/namespaces/snowlab/jobs et GET /api/v1/namespaces/snowlab/jobs/<job>/runs (état, début, datasets de sortie de chaque exécution). Équivalent Snowflake : lineage de Snowsight (https://docs.snowflake.com/en/user-guide/ui-snowsight-lineage, lue le 2026-10-23) : mouvements de données par COPY INTO, CREATE TABLE AS SELECT, INSERT ... SELECT, MERGE et UPDATE qui lisent une autre table, dépendances de vues, Enterprise Edition requise. Snowflake construit ce lineage à partir de son historique de requêtes : rien à émettre côté client. Le lineage des tables du client reste à constater chez lui.

Schémas `shared_*` (Marketplace et Data Sharing simulés)

ingest/shared_snapshots.py :

shared_calendar.calendar_days : snapshot_date, calendar_date, name, kind (public_holiday, school_holiday), region, _loaded_at.
write_calendar_snapshot(catalog, snapshot_date, horizon_days=730) écrit le calendrier visible à cette date (de un an avant à horizon_days après). Une date de snapshot ne s'écrit qu'une fois (SnapshotExistsError) ; seuls les schémas préfixés shared_ sont acceptés.
Seul le « fournisseur » (ce module, via lab_tech et PyIceberg) écrit. Les consommateurs sont en lecture seule : règle Trino SELECT seulement sur shared_* (y compris pour lab_tech), privilèges Polaris en lecture pour lab_reader.

Équivalent Snowflake : Data Sharing et Marketplace (https://docs.snowflake.com/en/user-guide/data-sharing-intro, lue le 2026-10-23) : les objets partagés sont en lecture seule, le consommateur crée une base en lecture seule à partir du partage, les tables Iceberg sont partageables, la Marketplace passe par des listings. Conforme à la simulation en lecture seule des schémas shared_*.