In de vorige twee delen hebben we het live databasemodel gepresenteerd voor een op abonnementen gebaseerd bedrijf en een datawarehouse (DWH) dat we zouden kunnen gebruiken voor rapportage. Hoewel het duidelijk is dat ze moeten samenwerken, was er geen verband tussen deze twee modellen. Vandaag nemen we de volgende stap en schrijven we de code om gegevens uit de live database over te zetten naar onze DWH.
De gegevensmodellen
Voordat we in de code duiken, laten we onszelf herinneren aan de twee modellen waarmee we zullen werken. De eerste is het transactiegegevensmodel dat we zullen gebruiken om onze realtime gegevens op te slaan. Aangezien we een op abonnementen gebaseerd bedrijf hebben, moeten we klant- en abonnementsgegevens, bestellingen van klanten en de bestelstatussen opslaan.
Er is echt veel dat we aan dit model kunnen toevoegen, zoals het volgen van betalingen en het opslaan van historische gegevens (vooral wijzigingen in klant- en abonnementsgegevens). Om het ETL-proces (extraheren, transformeren en laden) te benadrukken, wil ik dit model echter zo eenvoudig mogelijk houden.
Het gebruik van een transactiegegevensmodel als rapportagedatabase kan in sommige gevallen werken, maar niet in alle gevallen. We hebben dat al genoemd, maar het is de moeite waard om het te herhalen. Als we onze rapportagetaken willen scheiden van onze realtime processen, moeten we een soort rapportagedatabase creëren. Een datawarehouse is een oplossing.
Onze DWH is gecentreerd rond vier feitentabellen. De eerste twee houden het aantal klanten en abonnementen op dagniveau bij. De overige twee volgen het aantal leveringen en de producten in deze leveringen.
Mijn veronderstelling is dat we ons ETL-proces één keer per dag zullen uitvoeren. Eerst vullen we dimensietabellen met nieuwe waarden (indien nodig). Daarna vullen we feitentabellen.
Om onnodige herhalingen te voorkomen, zal ik alleen de code demonstreren die de eerste twee dimensietabellen en de eerste twee feitentabellen zal vullen. De overige tabellen kunnen worden gevuld met zeer vergelijkbare code. Ik raad je aan om de code zelf op te schrijven. Er is geen betere manier om iets nieuws te leren dan door het te proberen.
Het idee:dimensietabellen
Het algemene idee is om opgeslagen procedures te creëren die we regelmatig zouden kunnen gebruiken om de DWH te vullen -- zowel dimensietabellen als feitentabellen. Met deze procedures worden gegevens tussen twee databases op dezelfde server overgedragen. Dit betekent dat sommige query's binnen deze procedures tabellen uit beide databases zullen gebruiken. Dit wordt verwacht; we moeten de status van de DWH vergelijken met de live DB en wijzigingen aanbrengen in de DWH op basis van wat er in de live DB gebeurt.
We hebben vier dimensietabellen in onze DWH:dim_time , dim_city , dim_product , en dim_delivery_status .
De tijddimensie wordt ingevuld door de vorige datum toe te voegen. De belangrijkste veronderstelling is dat we deze procedure dagelijks uitvoeren, na sluitingstijd.
De afmetingen van de stad en het product zijn afhankelijk van de huidige waarden die zijn opgeslagen in de city en product woordenboeken in de live database. Als we iets aan deze woordenboeken toevoegen, dan worden bij de volgende DWH-update nieuwe waarden toegevoegd aan de dimensietabellen.
De laatste dimensietabel is de dim_delivery_status tafel. Het wordt niet bijgewerkt omdat het slechts drie standaardwaarden bevat. Een levering is onderweg, geannuleerd of afgeleverd.
Het idee:feitentabellen
Feitentabellen vullen is eigenlijk het echte werk. Hoewel de woordenboeken in de live database geen tijdstempelattribuut bevatten, doen tabellen met gegevens die zijn ingevoegd als resultaat van onze bewerkingen dat wel. U ziet twee tijdstempelkenmerken, time_inserted en time_updated , in het datamodel.
Nogmaals, ik ga ervan uit dat we de DWH-import eenmaal per dag met succes zullen uitvoeren. Dit stelt ons in staat om de gegevens op dagniveau te aggregeren. We tellen het aantal actieve en opgezegde klanten en abonnementen, evenals de leveringen en geleverde producten voor die datum.
Ons live-model werkt goed als we een invoegprocedure uitvoeren na de COB (close of business). Als we echter meer flexibiliteit willen, moeten we enkele wijzigingen in het model aanbrengen. Een dergelijke wijziging zou kunnen zijn dat er een aparte geschiedenistabel is om het exacte moment bij te houden waarop gegevens met betrekking tot klanten of abonnementen zijn gewijzigd. Met onze huidige organisatie weten we dat de wijziging heeft plaatsgevonden, maar we weten niet of er vóór deze wijziging heeft plaatsgevonden (bijv. een klant heeft gisteren opgezegd, zijn account na middernacht opnieuw geactiveerd en vandaag weer opgezegd) .
Dimensietabellen invullen
Zoals eerder vermeld, ga ik ervan uit dat we de DWH-import precies één keer per dag uitvoeren. Als dat niet het geval is, hebben we aanvullende code nodig om nieuw ingevoegde gegevens uit de dimensie- en feitentabellen te verwijderen. Voor de dimensietabellen zou dit beperkt zijn tot het verwijderen van de opgegeven datum.
Eerst controleren we of de opgegeven datum bestaat in de dim_time tafel. Zo niet, dan voegen we een nieuwe rij toe aan de tabel; als dat zo is, hoeven we niets te doen. In de meeste gevallen worden alle datums ingevoegd tijdens de eerste productie-implementatie. Maar ik ga met dit voorbeeld mee voor educatieve doeleinden.
Voor de dim_city en dim_product dimensies, voeg ik alleen nieuwe waarden toe die ik detecteer in de city en product tafels. Ik zal geen verwijderingen maken omdat naar eerder ingevoegde waarden kan worden verwezen in een feitentabel. We zouden kunnen gaan voor een zachte verwijdering, b.v. met een “actieve” vlag die we aan en uit konden zetten.
Voor de laatste tafel, dim_delivery_status , ik zal niets doen omdat het altijd dezelfde drie waarden zal bevatten.
De onderstaande code creëert een procedure die de dimensietabellen dim_time en dim_city .
Voor de tijddimensie voeg ik de datum van gisteren toe. Ik ga ervan uit dat het ETL-proces direct na middernacht begint. Ik controleer of die dimensie al bestaat en zo niet, dan voeg ik de nieuwe datum toe aan de tabel.
Voor de stadsdimensie gebruik ik een LEFT JOIN om gegevens uit de live database en de DWH-database samen te voegen om te bepalen welke rijen ontbreken. Dan voeg ik alleen ontbrekende gegevens toe aan de dimensietabel. Het is vermeldenswaard dat er een paar manieren zijn om te controleren of gegevens zijn gewijzigd. Dit proces wordt change data capture of CDC genoemd. Een veelgebruikte methode is het controleren op bijgewerkte tijdstempels of versies. Er zijn een paar extra manieren, maar deze vallen buiten het bestek van dit artikel.
Laten we nu eens kijken naar de code, die is geschreven met MySQL-syntaxis .
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 ()
Deze procedure uitvoeren -- wat we doen met behulp van de becommentarieerde procedure CALL -- voegt een nieuwe datum en alle ontbrekende steden toe aan de dimensietabellen. Probeer uw eigen code toe te voegen om de resterende twee dimensietabellen te vullen met nieuwe waarden.
Het ETL-proces in een datawarehouse
Het belangrijkste idee achter datawarehousing is om geaggregeerde gegevens in het gewenste formaat te bevatten. Natuurlijk moeten we dat formaat kennen voordat we zelfs maar beginnen met het bouwen van het magazijn. Als we alles volgens plan hebben gedaan, kunnen we alle voordelen krijgen die een DWH ons biedt. Het belangrijkste voordeel zijn verbeterde prestaties bij het uitvoeren van query's. Onze zoekopdrachten werken met minder records (omdat ze geaggregeerd zijn) en worden uitgevoerd in de rapportagedatabase (in plaats van in de live).
Maar voordat we een vraag kunnen stellen, moeten we feiten in onze database opslaan. De manier waarop we dat doen, hangt af van wat we later met onze gegevens moeten doen. Als we geen goed totaalbeeld hebben voordat we beginnen met het bouwen van onze DWH, kunnen we snel in de problemen komen! binnenkort.
De naam van dit proces is ETL:E =Extract, T =Transform, L =Load. Het grijpt de gegevens, transformeert deze naar de DWH-structuur en laadt deze in de DWH. Om precies te zijn, het daadwerkelijke proces dat we zullen gebruiken is ELT:Extract, Load, Transform. Omdat we opgeslagen procedures gebruiken, extraheren we gegevens, laden ze en transformeren we ze om aan onze behoeften te voldoen. Het is goed om te weten dat hoewel ETL en ELT enigszins van elkaar verschillen, de termen soms door elkaar worden gebruikt.
De feitentabellen invullen
Het vullen van feitentabellen is waarom we hier echt zijn. Vandaag vul ik twee feitentabellen, de fact_customer_subscribed tabel en de fact_subscription_status tafel. De overige twee feitentabellen kunt u als huiswerk uitproberen.
Voordat we verder gaan met het vullen van de feitentabel, moeten we aannemen dat de dimensietabellen zijn gevuld met nieuwe waarden. Het invullen van de feitentabellen volgt hetzelfde patroon. Omdat ze dezelfde structuur hebben, zal ik ze allebei samen uitleggen.
We groeperen gegevens op twee dimensies:tijd en stad. De tijddimensie wordt ingesteld op gisteren en we vinden de ID van het gerelateerde record in de dim_time tabel door datums te vergelijken (de laatste INNER JOIN in beide zoekopdrachten).
De ID van dim_city wordt geëxtraheerd door alle attributen samen te voegen die een UNIEKE combinatie vormen in de dimensietabel (plaatsnaam, postcode en landnaam).
In deze query testen we waarden met CASE en SOM ze vervolgens. Voor actieve en inactieve klanten heb ik de datum niet getest. Ik heb echter de huidige waarden voor deze velden geselecteerd. Voor nieuwe en geannuleerde accounts heb ik de bijgewerkte tijd getest.
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 ()
Nogmaals, ik heb commentaar gegeven op de laatste regel. Verwijder de opmerking en u kunt deze regel gebruiken om de procedure aan te roepen en nieuwe waarden in te voegen. Houd er rekening mee dat ik geen bestaande oude waarden heb verwijderd, dus deze procedure werkt niet als we al waarden hebben voor die datum en stad. Dit kan worden opgelost door verwijderingen uit te voeren vóór invoegingen.
Onthoud dat we de resterende feitentabellen in onze DWH moeten invullen. Ik moedig je aan om dat zelf te proberen!
Een ander ding dat ik zeker zou aanraden, is om het hele proces in een transactie te plaatsen. Dat zou ervoor zorgen dat ofwel alle invoegingen slagen of dat er geen worden gemaakt. Dit is erg belangrijk wanneer we willen voorkomen dat gegevens gedeeltelijk worden ingevoegd, b.v. als we meerdere procedures hebben voor het invoegen van dimensies en feiten en sommige doen hun werk terwijl andere falen.
Wat denk je?
Vandaag hebben we gezien hoe we het ELT/ETL-proces konden uitvoeren en gegevens uit een live database in een datawarehouse konden laden. Hoewel het proces dat we hebben gedemonstreerd behoorlijk vereenvoudigd is, bevat het alle elementen die nodig zijn om de gegevens te E(extraheren), T(om te zetten) in een geschikt formaat en tenslotte L(oad) het in het DWH. Wat denk je? Vertel ons uw ervaringen in de opmerkingen hieronder.