I de foregående to dele har vi præsenteret live-databasemodellen for en abonnementsbaseret virksomhed og et datavarehus (DWH), vi kunne bruge til rapportering. Selvom det er indlysende, at de burde arbejde sammen, var der ingen forbindelse mellem disse to modeller. I dag tager vi det næste trin og skriver koden til at overføre data fra livedatabasen til vores DWH.
Datamodellerne
Før vi dykker ned i koden, lad os minde os selv om de to modeller, vi vil arbejde med. Først er den transaktionsdatamodel, vi vil bruge til at gemme vores realtidsdata. I betragtning af, at vi driver en abonnementsbaseret virksomhed, skal vi gemme kunde- og abonnementsoplysninger, kunders ordrer og ordrestatusser.
Der er virkelig meget, vi kunne tilføje til denne model, såsom sporing af betalinger og lagring af historiske data (især ændringer i kunde- og abonnementsdata). For at understrege ETL-processen (ekstrahere, transformere og indlæse) ønsker jeg dog at holde denne model så enkel som muligt.
Brug af en transaktionsdatamodel som en rapporteringsdatabase fungerer muligvis i nogle tilfælde, men det vil ikke fungere i alle tilfælde. Vi har allerede nævnt det, men det er umagen værd at gentage det. Hvis vi ønsker at adskille vores rapporteringsopgaver fra vores realtidsprocesser, bør vi oprette en form for rapporteringsdatabase. Et datavarehus er én løsning.
Vores DWH er centreret omkring fire faktatabeller. De to første sporer antallet af kunder og abonnementer på dagligt niveau. De resterende to sporer antallet af leverancer og de produkter, der er inkluderet i disse leverancer.
Min antagelse er, at vi kører vores ETL-proces én gang om dagen. Først udfylder vi dimensionstabeller med nye værdier (hvor det er nødvendigt). Derefter udfylder vi faktatabeller.
For at undgå unødvendige gentagelser vil jeg kun demonstrere koden, der vil udfylde de første to dimensionstabeller og de to første faktatabeller. De resterende tabeller kan udfyldes med meget lignende kode. Jeg opfordrer dig til selv at skrive koden ned. Der er ingen bedre måde at lære noget nyt på end ved at prøve det.
Idéen:Dimensionstabeller
Den generelle idé er at skabe lagrede procedurer, som vi regelmæssigt kunne bruge til at udfylde DWH -- dimensionstabeller såvel som faktatabeller. Disse procedurer vil overføre data mellem to databaser på den samme server. Det betyder, at nogle forespørgsler i disse procedurer vil bruge tabeller fra begge databaser. Dette forventes; vi er nødt til at sammenligne DWH'ens tilstand med den levende DB og foretage ændringer i DWH i henhold til, hvad der sker i den levende DB.
Vi har fire dimensionstabeller i vores DWH:dim_time , dim_city , dim_product og dim_delivery_status .
Tidsdimensionen udfyldes ved at tilføje den forrige dato. Hovedantagelsen er, at vi kører denne procedure dagligt efter lukketid.
By- og produktdimensionerne vil afhænge af de aktuelle værdier, der er gemt i city og product ordbøger i live-databasen. Hvis vi tilføjer noget til disse ordbøger, vil nye værdier blive tilføjet til dimensionstabellerne ved den næste DWH-opdatering.
Den sidste dimensionstabel er dim_delivery_status bord. Det bliver ikke opdateret, fordi det kun indeholder tre standardværdier. En levering er enten i transit, annulleret eller leveret.
Idéen:Faktatabeller
At udfylde faktatabeller er faktisk det rigtige job. Selvom ordbøgerne i livedatabasen ikke indeholder en tidsstempelattribut, gør tabeller med data indsat som et resultat af vores operationer. Du vil bemærke to tidsstempelattributter, time_inserted og time_updated , i datamodellen.
Igen antager jeg, at vi med succes vil køre DWH-importen en gang om dagen. Dette gør det muligt for os at aggregere dataene på et dagligt niveau. Vi tæller antallet af aktive og annullerede kunder og abonnementer samt leveringer og leverede produkter for den dato.
Vores live-model fungerer godt, hvis vi kører en indsættelsesprocedure efter COB (afslutning). Alligevel, hvis vi ønsker mere fleksibilitet, bør vi foretage nogle ændringer i modellen. En sådan ændring kunne være at have en separat historiktabel til at spore det nøjagtige tidspunkt, hvor data relateret til kunder eller abonnementer ændrede sig. Med vores nuværende organisation ved vi, at ændringen er sket, men vi ved ikke, om der var nogen ændringer før denne (f.eks. annullerede en kunde i går, genaktiverede sin konto efter midnat og annullerede derefter igen i dag) .
Befolkning af dimensionstabeller
Som nævnt før, vil jeg gå ud fra den antagelse, at vi kører DWH-importen nøjagtigt en gang om dagen. Hvis det ikke er tilfældet, har vi brug for yderligere kode for at slette nyligt indsatte data fra dimensions- og faktatabellerne. For dimensionstabellerne vil dette være begrænset til sletning af den givne dato.
Først vil vi kontrollere, om den givne dato findes i dim_time bord. Hvis ikke, tilføjer vi en ny række til tabellen; hvis det gør, behøver vi ikke at gøre noget. I de fleste tilfælde indsættes alle datoer under den indledende produktionsinstallation. Men jeg vil gå med dette eksempel til uddannelsesformål.
Til dim_city og dim_product dimensioner, tilføjer jeg kun de nye værdier, jeg finder i city og product tabeller. Jeg vil ikke foretage nogen sletninger, fordi eventuelle tidligere indsatte værdier kunne henvises til i en faktatabel. Vi kunne gå med en blød sletning, f.eks. at have et "aktivt" flag, som vi kunne tænde og slukke for.
For den sidste tabel, dim_delivery_status , jeg vil ikke gøre noget, fordi det altid vil indeholde de samme tre værdier.
Koden nedenfor opretter en procedure, der vil udfylde dimensionstabellerne dim_time og dim_city .
For tidsdimensionen tilføjer jeg gårsdagens dato. Jeg går ud fra den antagelse, at ETL-processen starter lige efter midnat. Jeg tjekker, om den dimension allerede eksisterer, og hvis ikke, tilføjer jeg den nye dato i tabellen.
For bydimensionen bruger jeg en LEFT JOIN til at forbinde data fra live-databasen og DWH-databasen for at bestemme, hvilke rækker der mangler. Så tilføjer jeg kun eventuelle manglende data til dimensionstabellen. Det er værd at nævne, at der er et par måder at kontrollere, om data er blevet ændret på. Denne proces kaldes change data capture eller CDC. En almindelig metode er at søge efter opdaterede tidsstempler eller versioner. Der er få yderligere måder, men de er uden for denne artikels omfang.
Lad os tage et kig på koden nu, som er skrevet ved hjælp af MySQL-syntaks .
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 ()
Kører denne procedure -- hvilket vi gør ved at bruge den kommenterede procedure CALL -- indsætter en ny dato og alle manglende byer i dimensionstabellerne. Prøv at tilføje din egen kode for at udfylde de resterende to dimensionstabeller med nye værdier.
ETL-processen i et datavarehus
Hovedideen bag data warehousing er at indeholde aggregerede data i det ønskede format. Selvfølgelig skal vi kende det format, før vi overhovedet begynder at bygge lageret. Hvis vi har gjort alt som planlagt, kan vi få alle de fordele, en DWH tilbyder os. Den største fordel er forbedret ydeevne, når du kører forespørgsler. Vores forespørgsler arbejder med færre poster (fordi de er aggregerede) og kører på rapporteringsdatabasen (i stedet for den levende).
Men før vi kan forespørge, skal vi gemme fakta i vores database. Den måde, vi gør det på, afhænger af, hvad vi skal gøre med vores data senere. Hvis vi ikke har et godt overordnet billede, før vi begynder at bygge vores DWH, kan vi hurtigt komme i problemer! snart.
Navnet på denne proces er ETL:E =Extract, T =Transform, L =Load. Den griber dataene, transformerer dem, så de passer til DWH-strukturen, og indlæser dem i DWH. For at være præcis er den faktiske proces, vi vil bruge, ELT:Udtræk, indlæs, transformer. Da vi bruger lagrede procedurer, udtrækker vi data, indlæser dem og transformerer dem derefter for at opfylde vores behov. Det er godt at vide, at selvom ETL og ELT er lidt forskellige, bruges udtrykkene nogle gange i flæng.
Population af faktatabellerne
At udfylde faktatabeller er derfor, vi virkelig er her. I dag vil jeg udfylde to faktatabeller, fact_customer_subscribed tabellen og fact_subscription_status bord. De resterende to faktatabeller kan du prøve som hjemmearbejde.
Før vi går videre til at udfylde faktatabel, må vi antage, at dimensionstabellerne er udfyldt med nye værdier. At udfylde faktatabellerne følger det samme mønster. Da de har samme struktur, vil jeg forklare dem begge sammen.
Vi grupperer data efter to dimensioner:tid og by. Tidsdimensionen indstilles til i går, og vi finder ID'et for den relaterede post i dim_time tabel ved at sammenligne datoer (den sidste INNER JOIN i begge forespørgsler).
ID'et for dim_city udtrækkes ved at samle alle de attributter, der danner en UNIK kombination i dimensionstabellen (bynavn, postnummer og landenavn).
I denne forespørgsel tester vi værdier med CASE og SUMMER dem derefter. For aktive og inaktive kunder har jeg ikke testet datoen. Jeg har dog valgt, som de er, værdier for disse felter. For nye og annullerede konti har jeg testet den opdaterede tid.
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 ()
Igen har jeg kommenteret den sidste linje. Fjern kommentaren, og du kan bruge denne linje til at kalde proceduren og indsætte nye værdier. Bemærk venligst, at jeg ikke har slettet nogen eksisterende gamle værdier, så denne procedure vil ikke fungere, hvis vi allerede har værdier for denne dato og by. Dette kan løses ved at udføre sletninger før indsættelser.
Husk, vi skal udfylde de resterende faktatabeller i vores DWH. Jeg opfordrer dig til at prøve det selv!
En anden ting, jeg klart vil anbefale, er at placere hele processen i en transaktion. Det ville sikre, at enten alle indsættelser lykkes, eller at ingen foretages. Det er meget vigtigt, når vi vil undgå at få data delvist indsat, f.eks. hvis vi har flere procedurer til at indsætte dimensioner og fakta, og nogle af dem gør deres arbejde, mens andre fejler.
Hvad synes du?
I dag har vi set, hvordan vi kunne udføre ELT/ETL-processen og indlæse data fra en live database til et datavarehus. Selvom den proces, vi demonstrerede, er temmelig forenklet, indeholder den alle de nødvendige elementer for at E(ekstrahere) dataene, T(omformere) dem til et passende format og til sidst L(oad) dem ind i DWH. Hvad synes du? Fortæl os venligst dine oplevelser i kommentarerne nedenfor.