Dans les deux parties précédentes, nous avons présenté le modèle de base de données en direct pour une entreprise basée sur un abonnement et un entrepôt de données (DWH) que nous pourrions utiliser pour les rapports. Bien qu'il soit évident qu'ils devraient fonctionner ensemble, il n'y avait aucun lien entre ces deux modèles. Aujourd'hui, nous allons passer à l'étape suivante et écrire le code pour transférer les données de la base de données en direct vers notre DWH.
Les modèles de données
Avant de plonger dans le code, rappelons-nous les deux modèles avec lesquels nous allons travailler. Le premier est le modèle de données transactionnel que nous utiliserons pour stocker nos données en temps réel. Étant donné que nous gérons une entreprise par abonnement, nous devrons stocker les détails des clients et des abonnements, les commandes des clients et les statuts des commandes.
Il y a vraiment beaucoup de choses que nous pourrions ajouter à ce modèle, comme le suivi des paiements et le stockage des données historiques (en particulier les changements dans les données des clients et des abonnements). Pour mettre l'accent sur le processus ETL (extraction, transformation et chargement), cependant, je souhaite que ce modèle reste aussi simple que possible.
L'utilisation d'un modèle de données transactionnel comme base de données de rapports peut fonctionner dans certains cas, mais cela ne fonctionnera pas dans tous les cas. Nous l'avons déjà mentionné, mais cela vaut la peine de le répéter. Si nous voulons séparer nos tâches de reporting de nos processus en temps réel, nous devons créer une sorte de base de données de reporting. Un entrepôt de données est une solution.
Notre DWH est centré autour de quatre tables de faits. Les deux premiers suivent le nombre de clients et d'abonnements au niveau quotidien. Les deux autres suivent le nombre de livraisons et les produits inclus dans ces livraisons.
Mon hypothèse est que nous exécuterons notre processus ETL une fois par jour. Tout d'abord, nous allons remplir les tables de dimension avec de nouvelles valeurs (si nécessaire). Après cela, nous remplirons les tables de faits.
Pour éviter les répétitions inutiles, je ne montrerai que le code qui remplira les deux premières tables de dimension et les deux premières tables de faits. Les tables restantes peuvent être remplies à l'aide d'un code très similaire. Je vous encourage à écrire le code vous-même. Il n'y a pas de meilleur moyen d'apprendre quelque chose de nouveau qu'en l'essayant.
L'idée :tableaux de dimensions
L'idée générale est de créer des procédures stockées que nous pourrions utiliser régulièrement pour remplir le DWH - tables de dimension ainsi que des tables de faits. Ces procédures transféreront des données entre deux bases de données sur le même serveur. Cela signifie que certaines requêtes à l'intérieur de ces procédures utiliseront des tables des deux bases de données. C'est prévu; nous devons comparer l'état du DWH à la base de données en direct et apporter des modifications au DWH en fonction de ce qui se passe dans la base de données en direct.
Nous avons quatre tables de dimensions dans notre DWH :dim_time , dim_city , dim_product , et dim_delivery_status .
La dimension temporelle est renseignée en ajoutant la date précédente. L'hypothèse principale est que nous exécuterons cette procédure quotidiennement, après la fermeture des bureaux.
Les dimensions de la ville et du produit dépendront des valeurs actuelles stockées dans le city et product dictionnaires dans la base de données en direct. Si nous ajoutons quelque chose à ces dictionnaires, de nouvelles valeurs seront ajoutées aux tables de dimension lors de la prochaine mise à jour de DWH.
La dernière table de dimension est le dim_delivery_status table. Il ne sera pas mis à jour car il ne contient que trois valeurs par défaut. Une livraison est en transit, annulée ou livrée.
L'idée :les tableaux de faits
Le remplissage des tables de faits est en fait le vrai travail. Alors que les dictionnaires de la base de données en direct ne contiennent pas d'attribut d'horodatage, les tables avec des données insérées à la suite de nos opérations en contiennent. Vous remarquerez deux attributs d'horodatage, time_inserted et time_updated , dans le modèle de données.
Encore une fois, je suppose que nous exécuterons avec succès l'importation DWH une fois par jour. Cela nous permet d'agréger les données au niveau quotidien. Nous comptabiliserons le nombre de clients et d'abonnements actifs et annulés, ainsi que les livraisons et les produits livrés à cette date.
Notre modèle en direct fonctionne bien si nous exécutons une procédure d'insertion après le COB (close of business). Néanmoins, si nous voulons plus de flexibilité, nous devrions apporter quelques modifications au modèle. L'un de ces changements pourrait être d'avoir une table d'historique distincte pour suivre le moment exact où les données relatives aux clients ou aux abonnements ont changé. Avec notre organisation actuelle, nous saurons que le changement s'est produit, mais nous ne saurons pas s'il y a eu des changements avant celui-ci (par exemple, un client a annulé hier, a réactivé son compte après minuit, puis a annulé à nouveau aujourd'hui) .
Remplir les tableaux de dimensions
Comme mentionné précédemment, je partirai du principe que nous exécuterons l'importation DWH exactement une fois par jour. Si ce n'est pas le cas, nous aurions besoin de code supplémentaire pour supprimer les données nouvellement insérées des tables de dimension et de faits. Pour les tables de dimension, cela se limiterait à supprimer la date donnée.
Tout d'abord, nous allons vérifier si la date donnée existe dans le dim_time table. Sinon, nous ajouterons une nouvelle ligne au tableau ; si c'est le cas, nous n'avons rien à faire. Dans la plupart des cas, toutes les dates sont insérées lors du déploiement initial en production. Mais je vais utiliser cet exemple à des fins éducatives.
Pour le dim_city et dim_product dimensions, je n'ajouterai que les nouvelles valeurs que je détecte dans la city et product les tables. Je ne ferai aucune suppression car toutes les valeurs précédemment insérées pourraient être référencées dans une table de faits. Nous pourrions opter pour une suppression douce, par ex. avoir un indicateur "actif" que nous pourrions activer et désactiver.
Pour le dernier tableau, dim_delivery_status , je ne ferai rien car il contiendra toujours les trois mêmes valeurs.
Le code ci-dessous crée une procédure qui remplira les tables de dimension dim_time et dim_city .
Pour la dimension temporelle, j'ajouterai la date d'hier. Je pars du principe que le processus ETL commence juste après minuit. Je vais vérifier si cette dimension existe déjà et si ce n'est pas le cas, j'ajouterai la nouvelle date dans le tableau.
Pour la dimension de la ville, j'utiliserai un LEFT JOIN pour joindre les données de la base de données en direct et de la base de données DWH afin de déterminer quelles lignes manquent. Ensuite, j'ajouterai uniquement les données manquantes à la table de dimension. Il convient de mentionner qu'il existe plusieurs façons de vérifier si les données ont été modifiées. Ce processus est appelé capture de données modifiées, ou CDC. Une méthode courante consiste à vérifier les horodatages ou les versions mis à jour. Il existe quelques moyens supplémentaires, mais ils sortent du cadre de cet article.
Jetons un coup d'œil au code maintenant, qui est écrit en utilisant la syntaxe MySQL .
DROP PROCEDURE IF EXISTS p_update_dimensions//
CREATE PROCEDURE p_update_dimensions ()
BEGIN
SET @time_exists = 0;
SET @time_date = DATE_ADD(DATE(NOW()), INTERVAL -1 DAY);
-- procedure populates dimension tables with new values
-- dim_time
SET @time_exists = (SELECT COUNT(*) FROM subscription_dwh.dim_time dim_time WHERE dim_time.time_date = @time_date);
IF (@time_exists = 0) THEN
INSERT INTO subscription_dwh.`dim_time`(`time_date`, `time_year`, `time_month`, `time_week`, `time_weekday`, `ts`)
SELECT
@time_date AS time_date,
YEAR(@time_date) AS time_year,
MONTH(@time_date) AS time_month,
WEEK(@time_date) AS time_week,
WEEKDAY(@time_date) AS time_weekday,
NOW() AS ts;
END IF;
-- dim_city
INSERT INTO subscription_dwh.`dim_city`(`city_name`, `postal_code`, `country_name`, `ts`)
SELECT
city_live.city_name,
city_live.postal_code,
country_live.country_name,
Now()
FROM subscription_live.city city_live
INNER JOIN subscription_live.country country_live
ON city_live.country_id = country_live.id
LEFT JOIN subscription_dwh.dim_city city_dwh
ON city_live.city_name = city_dwh.city_name
AND city_live.postal_code = city_dwh.postal_code
AND country_live.country_name = city_dwh.country_name
WHERE city_dwh.id IS NULL;
END//
-- CALL p_update_dimensions ()
Exécuter cette procédure -- ce que nous faisons en utilisant la procédure commentée CALL -- insère une nouvelle date et toutes les villes manquantes dans les tables de dimension. Essayez d'ajouter votre propre code pour remplir les deux tables de dimension restantes avec de nouvelles valeurs.
Le processus ETL dans un entrepôt de données
L'idée principale derrière l'entreposage de données est de contenir des données agrégées dans le format souhaité. Bien sûr, nous devrions connaître ce format avant même de commencer à construire l'entrepôt. Si nous avons tout fait comme prévu, nous pouvons obtenir tous les avantages qu'un DWH nous offre. Le principal avantage est l'amélioration des performances lors de l'exécution des requêtes. Nos requêtes fonctionnent avec moins d'enregistrements (car ils sont agrégés) et s'exécutent sur la base de données de rapports (plutôt que sur celle en direct).
Mais avant de pouvoir interroger, nous devons stocker des faits dans notre base de données. La façon dont nous procéderons dépendra de ce que nous devrons faire de nos données plus tard. Si nous n'avons pas une bonne vue d'ensemble avant de commencer à construire notre DWH, nous pourrions bientôt nous retrouver en difficulté ! bientôt.
Le nom de ce processus est ETL :E =Extraire, T =Transformer, L =Charger. Il saisit les données, les transforme pour s'adapter à la structure DWH et les charge dans le DWH. Pour être précis, le processus réel que nous utiliserons est ELT :Extract, Load, Transform. Puisque nous utilisons des procédures stockées, nous allons extraire des données, les charger, puis les transformer pour répondre à nos besoins. Il est bon de savoir que même si ETL et ELT sont légèrement différents, les termes sont parfois utilisés de manière interchangeable.
Remplir les tables de faits
Remplir les tables de faits est la raison pour laquelle nous sommes vraiment là. Aujourd'hui, je vais remplir deux tables de faits, le fact_customer_subscribed table et le fact_subscription_status table. Les deux tables de faits restantes sont à vous d'essayer comme devoirs.
Avant de passer au remplissage de la table de faits, nous devons supposer que les tables de dimension sont remplies avec de nouvelles valeurs. Le remplissage des tables de faits suit le même schéma. Puisqu'ils ont la même structure, je vais les expliquer ensemble.
Nous regroupons les données selon deux dimensions :le temps et la ville. La dimension temporelle sera définie sur hier et nous trouverons l'ID de l'enregistrement associé dans le dim_time table en comparant les dates (le dernier INNER JOIN dans les deux requêtes).
L'identifiant de dim_city est extrait en joignant tous les attributs qui forment une combinaison UNIQUE dans la table de dimension (nom de la ville, code postal et nom du pays).
Dans cette requête, nous allons tester les valeurs avec CASE puis les SUM. Pour les clients actifs et inactifs, je n'ai pas testé la date. Cependant, j'ai sélectionné des valeurs telles quelles pour ces champs. Pour les comptes nouveaux et annulés, j'ai testé l'heure mise à jour.
DROP PROCEDURE IF EXISTS p_update_facts//
CREATE PROCEDURE p_update_facts ()
BEGIN
SET @time_date = DATE_ADD(DATE(NOW()), INTERVAL -1 DAY);
-- procedure populates fact tables with new values
-- fact_customer_subscribed
INSERT INTO `fact_customer_subscribed`(`dim_city_id`, `dim_time_id`, `total_active`, `total_inactive`, `daily_new`, `daily_canceled`, `ts`)
SELECT
city_dwh.id AS dim_ctiy_id,
time_dwh.id AS dim_time_id,
SUM(CASE WHEN customer_live.active = 1 THEN 1 ELSE 0 END) AS total_active,
SUM(CASE WHEN customer_live.active = 0 THEN 1 ELSE 0 END) AS total_inactive,
SUM(CASE WHEN customer_live.active = 1 AND DATE(customer_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_new,
SUM(CASE WHEN customer_live.active = 0 AND DATE(customer_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_canceled,
MIN(NOW()) AS ts
FROM subscription_live.`customer` customer_live
INNER JOIN subscription_live.`city` city_live ON customer_live.city_id = city_live.id
INNER JOIN subscription_live.`country` country_live ON city_live.country_id = country_live.id
INNER JOIN subscription_dwh.dim_city city_dwh
ON city_live.city_name = city_dwh.city_name
AND city_live.postal_code = city_dwh.postal_code
AND country_live.country_name = city_dwh.country_name
INNER JOIN subscription_dwh.dim_time time_dwh ON time_dwh.time_date = @time_date
GROUP BY
city_dwh.id,
time_dwh.id;
-- fact_subscription_status
INSERT INTO `fact_subscription_status`(`dim_city_id`, `dim_time_id`, `total_active`, `total_inactive`, `daily_new`, `daily_canceled`, `ts`)
SELECT
city_dwh.id AS dim_ctiy_id,
time_dwh.id AS dim_time_id,
SUM(CASE WHEN subscription_live.active = 1 THEN 1 ELSE 0 END) AS total_active,
SUM(CASE WHEN subscription_live.active = 0 THEN 1 ELSE 0 END) AS total_inactive,
SUM(CASE WHEN subscription_live.active = 1 AND DATE(subscription_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_new,
SUM(CASE WHEN subscription_live.active = 0 AND DATE(subscription_live.time_updated) = @time_date THEN 1 ELSE 0 END) AS daily_canceled,
MIN(NOW()) AS ts
FROM subscription_live.`customer` customer_live
INNER JOIN subscription_live.`subscription` subscription_live ON subscription_live.customer_id = customer_live.id
INNER JOIN subscription_live.`city` city_live ON customer_live.city_id = city_live.id
INNER JOIN subscription_live.`country` country_live ON city_live.country_id = country_live.id
INNER JOIN subscription_dwh.dim_city city_dwh
ON city_live.city_name = city_dwh.city_name
AND city_live.postal_code = city_dwh.postal_code
AND country_live.country_name = city_dwh.country_name
INNER JOIN subscription_dwh.dim_time time_dwh ON time_dwh.time_date = @time_date
GROUP BY
city_dwh.id,
time_dwh.id;
END//
-- CALL p_update_facts ()
Encore une fois, j'ai commenté la dernière ligne. Supprimez le commentaire et vous pouvez utiliser cette ligne pour appeler la procédure et insérer de nouvelles valeurs. Veuillez noter que je n'ai supprimé aucune ancienne valeur existante, donc cette procédure ne fonctionnera pas si nous avons déjà des valeurs pour cette date et cette ville. Cela peut être résolu en effectuant des suppressions avant les insertions.
N'oubliez pas que nous devons remplir les tables de faits restantes dans notre DWH. Je vous encourage à essayer vous-même !
Une autre chose que je recommanderais certainement est de placer l'ensemble du processus dans une transaction. Cela garantirait que toutes les insertions réussissent ou qu'aucune n'est effectuée. Ceci est très important lorsque nous voulons éviter d'avoir des données partiellement insérées, par ex. si nous avons plusieurs procédures pour insérer des dimensions et des faits et que certaines d'entre elles font leur travail tandis que d'autres échouent.
Qu'en pensez-vous ?
Aujourd'hui, nous avons vu comment nous pouvions exécuter le processus ELT/ETL et charger les données d'une base de données en direct dans un entrepôt de données. Bien que le processus que nous avons démontré soit assez simplifié, il contient tous les éléments nécessaires pour E(extraire) les données, les T(transformer) dans un format approprié, et enfin les L(charger) dans le DWH. Qu'en penses-tu? Veuillez nous faire part de vos expériences dans les commentaires ci-dessous.
