BigQuery — des données fraîches sans mutations massives

Cet article présente une approche pour relever un défi bien connu des entrepôts de données : conserver des données fraîches et à jour sans recourir à des mutations à grande échelle.

Chez DoiT, nous accompagnons de nombreux clients dans la conception de systèmes bien architecturés et l'utilisation efficace des services cloud. Ce qui suit s'inspire d'un cas client réel.

Une entreprise SaaS propose à ses clients une plateforme analytique reposant sur BigQuery. Pour certains clients, les données arrivent sous forme de snapshot complet, comprenant tout l'historique et non uniquement les changements incrémentaux depuis la dernière mise à jour.

L'un de ces clients fournit un snapshot de 15 To, alors que les données réellement modifiées ne représentent que 0,1 %. Faute de pouvoir récupérer uniquement les changements incrémentaux, l'équipe devait trouver le moyen le plus fiable et le plus efficace, en performance comme en coût, de mettre à jour la table existante avec les nouvelles données du snapshot.

L'entreprise et son client ont défini un contrat de données pour que le pipeline réponde aux exigences métier et techniques :

• Les données entrantes constituent un snapshot complet sur une fenêtre glissante de deux ans, livré six fois par jour.
• Elles sont partitionnées par jour (taille de partition d'environ 20 Gio) et marquées d'un identifiant de snapshot (valeur incrémentale).
• Elles contiennent à la fois de nouvelles données (nouvelles clés) et des données modifiées (mises à jour de données existantes).
• L'expiration des partitions est configurée à 2 ans.
• Les enregistrements présents dans le snapshot actuel mais absents du nouveau sont conservés jusqu'à l'expiration de la partition.

L'équipe produit de l'entreprise SaaS a défini les exigences suivantes :

• Modélisation : deux tables contiennent les données existantes et les nouvelles données.

Les nouvelles données résident dans une table nommée staging, et les données existantes (celles exposées au client) dans une table appelée target.

• Unicité : les données de la table target sont uniques (pas de clés en double).
• Fraîcheur : la table target est rafraîchie dans l'heure suivant l'arrivée du snapshot.
• Disponibilité : l'utilisateur final reçoit toujours une réponse à sa requête.

Comme l'explique l'article du blog Google ci-dessous, BigQuery n'est pas conçu comme les bases OLTP transactionnelles, qui sont nativement performantes pour les mutations massives.

" BigQuery n'est pas la seule base OLAP à imposer des contraintes sur la fréquence des mutations, que ce soit explicitement (via des quotas) ou implicitement (avec une dégradation significative des performances) : ce type de bases est optimisé pour l'ingestion à grande échelle et les requêtes analytiques, et non pour le traitement transactionnel.

Par ailleurs, BigQuery permet de lire l'état d'une table à n'importe quel instant durant les sept jours précédents. Cette fenêtre de consultation impose de conserver des données déjà supprimées par l'utilisateur. Pour garantir des requêtes rentables et efficaces à grande échelle, BigQuery limite la fréquence des mutations via des quotas. " ( Performing large-scale mutations in BigQuery | Google Cloud Blog )

Un remplacement intégral, qui supprime l'ancien snapshot pour le remplacer par le nouveau, ne suffit pas : il faut conserver les enregistrements présents dans le snapshot actuel mais absents du dernier snapshot.

Il faut donc comparer les enregistrements entre les deux snapshots, ce que la simple approche par remplacement ne permet pas.

Pour illustrer les options de mutation et leurs différences, posons précisément le problème, avec la modélisation des données et la logique de mutation exactes.

Schéma de la table :

Un enregistrement d'une table (staging ou target) contient un attribute_id et une valeur pour un produit donné (product_id) dans un magasin donné (store_id) à une date précise, ainsi que la référence du snapshot dans lequel il existait (update_id).

Exemple d'enregistrement :

D'après les données reçues dans le snapshot ID 1 (update_id), la quantité totale (attribute_id = 1) d'un produit (product_id = 301865) dans un magasin (store_id = 2072) était de 20 le 21/04/2017.

Exemple de logique de mutation :

Voici un échantillon issu d'une partition (date donnée) provenant de deux snapshots distincts.

Table target après ingestion du snapshot n°1

Le snapshot n°2 contenait les changements suivants :

1. Bleu : enregistrements mis à jour (présents dans les deux snapshots), valeur modifiée.

2. Rouge : enregistrement historique (présent uniquement dans le snapshot n°1).

3. Vert — nouvel enregistrement (présent uniquement dans le snapshot n°2).

Table target après ingestion du snapshot n°2

BigQuery prenant en charge MERGE, fusionner la table de staging avec la table cible apparaît comme l'approche intuitive, selon la logique ci-dessous.

Cette logique MERGE définit les mutations à appliquer selon les différents cas, sur la base d'une clé composée des champs suivants : product_id, store_id, attribute_id.

Voici l'implémentation de la logique MERGE :

MERGE
nw-playground.demo.merge_target T
USING
nw-playground.demo.merge_staging S
ON
 T.product_Id = S.product_Id
 AND T.store_Id = S.store_Id
 AND T.attribute_Id = S.attribute_Id
 AND T.date = S.date
 WHEN MATCHED and T.date  >= '2017-04-14' and T.date <= '2017-04-26'
 THEN UPDATE SET T.attr_value = S.attr_value, T.update_id = S.update_id
 WHEN NOT MATCHED BY TARGET  and S.date >= '2017-04-14' and S.date <= '2017-04-26'
 THEN
INSERT
 (date,
   product_Id,
   store_Id,
   attribute_Id,
   attr_value,
   update_id)
VALUES
 (S.date, S.Product_id, S.store_Id, S.attribute_Id, S.attr_value, S.update_id);

Problème :

Comme indiqué plus haut, l'opération MERGE est très coûteuse dans BQ et présente des limites : elle requiert un shuffling et la mutation des données existantes.

Il faut donc trouver un moyen d'éviter le MERGE et la mutation des données.

Que faisons-nous ? Au lieu de fusionner les enregistrements des tables staging et target, nous procédons ainsi :

1. Ajout du nouveau snapshot à la table staging.
2. Déduplication des enregistrements similaires en ne conservant que le plus récent.

La déduplication est un processus qui supprime les enregistrements en double pour n'en laisser qu'un seul. Elle s'appuie sur une clé d'enregistrement permettant d'identifier les doublons et sur une logique déterminant lequel conserver (par exemple, l'enregistrement le plus récent). 3. Remplacement de la table staging par le résultat de la déduplication. 4. Remplacement de la table target par un CLONE de la table staging.

" CLONE Un clone de table est une copie légère et inscriptible d'une autre table (appelée table de base). Vous n'êtes facturé que pour le stockage des données du clone qui diffèrent de la table de base ; au départ, il n'y a donc aucun coût de stockage pour un clone de table . Hormis le modèle de facturation du stockage et quelques métadonnées additionnelles pour la table de base, un clone se comporte comme une table standard : on peut l'interroger, le copier, le supprimer, etc. Une fois créé, le clone est indépendant de la table de base. Les modifications apportées à la table de base ou au clone ne sont pas répercutées sur l'autre. " table-clones-intro

Comment procédons-nous ?

1. Ajout des données : avec la commande LOAD en mode append, la table de staging contient à la fois le snapshot actuel et le nouveau.
2. Déduplication : on déduplique les enregistrements similaires (même clé) en ne conservant que le plus récent. La déduplication n'affecte pas les enregistrements entièrement nouveaux ni ceux présents uniquement dans le snapshot actuel. Elle s'implémente avec les fonctions de fenêtre, qui calculent des valeurs sur un groupe de lignes, combinées à l'opération QUALIFY. La clause QUALIFY permet de filtrer les résultats d'une fonction de fenêtre (groupe de lignes).
3. Remplacement de staging : avec CREATE OR REPLACE à partir des résultats de la requête de déduplication.
4. Clonage de staging vers target : avec CREATE OR REPLACE TABLE ... CLONE.

Pourquoi cette approche ?

Elle présente plusieurs avantages par rapport à l'approche MERGE :

1. Réduction du shuffling : ajouter le nouveau snapshot à l'existant évite le shuffling entre deux tables.

2. Plus d'opération MERGE/JOIN : la déduplication s'appuie sur une fonction de fenêtre utilisant la clause PARTITION BY, qui découpe les lignes d'entrée en partitions distinctes sur lesquelles la fonction est évaluée indépendamment.

3. Pas de mutation : de nouvelles tables sont créées ou clonées, ce qui contourne les quotas de mutation de BigQuery.

Exemple de logique d'ajout et de déduplication :

Table de staging après l'ajout du nouveau snapshot (n°2)

Comme avec le résultat du MERGE, la table après déduplication contient les changements suivants :

1. Bleu — enregistrements mis à jour (présents dans les deux snapshots), valeur modifiée.

2. Rouge — anciens enregistrements à conserver (présents uniquement dans le snapshot n°1).

3. Vert — nouvel enregistrement (présent uniquement dans le snapshot n°2).

Table de staging après déduplication

À l'arrivée du snapshot suivant, le même processus s'applique.

Exemples de code

Le code ci-dessous crée une nouvelle table de staging à partir des résultats de la déduplication, en remplaçant la table existante grâce à la clause QUALIFY et à un numéro de ligne attribué à chaque enregistrement de clé identique. Le 1er enregistrement (rownum=1) est le plus récent.

CREATE OR REPLACE TABLE
 `nw-playground.demo.dedup_staging`
 (date DATE,    product_id INT64,
  store_id INT64,    attribute_id INT64,
  attr_value INT64,    update_id INT64)
PARTITION BY
 date
CLUSTER BY
 product_id,
 store_id AS
SELECT
 date,  product_id,  store_id,  attribute_id,  attr_value,  update_id
FROM
 `nw-playground.demo.dedup_staging`
WHERE
 date >= '2017-04-14' AND date <= '2017-04-26'
 QUALIFY ROW_NUMBER()
 OVER(PARTITION BY date, product_id, store_id, attribute_id
 ORDER BY update_id DESC ) = 1;

CREATE OR REPLACE TABLE
`nw-playground.nw_demo.dedup_target`
CLONE `nw-playground.nw_demo.dedup_staging`;

Les tests ont porté sur 7,8 milliards d'enregistrements (350 Gio) répartis sur 13 partitions, avec environ 3 M de différences (tous types de modifications) entre staging et target. Ils ont utilisé le modèle de facturation à la demande.

Dans le cas du MERGE, la table staging ne contenait que le nouveau snapshot et la table target l'ancien. À l'inverse, dans le cas de la déduplication, la table de staging contenait à la fois le snapshot existant et le nouveau, et la table target était un clone du 1er snapshot de staging.

Le tableau ci-dessous compare les résultats :

Synthèse comparative des détails d'exécution

La comparaison des plans d'exécution montre clairement que la durée et l'utilisation des slots du test MERGE sont deux fois supérieures à celles du DEDUPLICATE :

Durée et utilisation des slots de l'opération principale

Modèle de facturation à la demande

Le coût est similaire d'un test à l'autre, puisque le volume de données scanné est identique. En charge complète, il atteint environ 187 $ (6,25 $ par Tio) sur la base de 30 To pour 2 snapshots.

BigQuery Editions

Le nouveau modèle de facturation introduit par Google en avril 2023 repose sur une tarification à la capacité (paiement par slot/heure) et propose trois éditions. Pour estimer le coût d'une charge en production, on peut tabler sur un chargement complet de snapshot avec un facteur de 56 par rapport à notre test (730 partitions-jours sur 2 ans, soit 730/13 = 56).

L'estimation reste approximative, car différentes configurations et commitments peuvent s'appliquer. Elle illustre surtout l'écart entre les deux options.

Comparaison des estimations de coût entre les éditions de BigQuery (tarifs US)

Il existe plusieurs façons d'implémenter la déduplication (avec ROWNUM ou GROUP BY). La déduplication via ROWNUM offre des performances comparables à celles de la clause QUALIFY ; QUALIFY reste cependant une commande plus lisible. L'utilisation de GROUP BY a donné de moins bons résultats, identiques à ceux du Merge.

Une autre voie efficace consiste à utiliser les procédures stockées pour Apache Spark ; à suivre dans un prochain article…