W poprzednich dwóch częściach przedstawiliśmy model bazy danych na żywo dla biznesu opartego na subskrypcji oraz hurtowni danych (DWH), którą moglibyśmy wykorzystać do raportowania. Chociaż oczywiste jest, że powinny ze sobą współpracować, nie było żadnego związku między tymi dwoma modelami. Dzisiaj zrobimy kolejny krok i napiszemy kod, aby przenieść dane z aktywnej bazy danych do naszego DWH.
Modele danych
Zanim zagłębimy się w kod, przypomnijmy sobie dwa modele, z którymi będziemy pracować. Pierwszy to model danych transakcyjnych, którego użyjemy do przechowywania naszych danych w czasie rzeczywistym. Biorąc pod uwagę, że prowadzimy działalność opartą na subskrypcji, będziemy musieli przechowywać dane klientów i subskrypcji, zamówienia klientów oraz statusy zamówień.
Do tego modelu możemy dodać naprawdę wiele, np. śledzenie płatności i przechowywanie danych historycznych (zwłaszcza zmiany w danych klientów i subskrypcji). Aby jednak podkreślić proces ETL (wyodrębnianie, przekształcanie i ładowanie), chcę, aby ten model był tak prosty, jak to tylko możliwe.
Używanie transakcyjnego modelu danych jako bazy danych raportowania może działać w niektórych przypadkach, ale nie we wszystkich przypadkach. Wspomnieliśmy już o tym, ale warto to powtórzyć. Jeśli chcemy oddzielić nasze zadania raportowe od naszych procesów w czasie rzeczywistym, powinniśmy stworzyć coś w rodzaju bazy danych raportowania. Hurtownia danych to jedno rozwiązanie.
Nasze DWH skupia się wokół czterech tabel faktów. Pierwsze dwa śledzą liczbę klientów i subskrypcji na poziomie dziennym. Pozostałe dwa śledzą liczbę dostaw i produkty zawarte w tych dostawach.
Wychodzę z założenia, że będziemy uruchamiać nasz proces ETL raz dziennie. Najpierw wypełnimy tabele wymiarów nowymi wartościami (w razie potrzeby). Następnie wypełnimy tabele faktów.
Aby uniknąć niepotrzebnych powtórzeń, zademonstruję tylko kod, który wypełni dwie pierwsze tabele wymiarów i dwie pierwsze tabele faktów. Pozostałe tabele można wypełnić bardzo podobnym kodem. Zachęcam do samodzielnego spisania kodu. Nie ma lepszego sposobu na nauczenie się czegoś nowego niż wypróbowanie tego.
Pomysł:Tabele wymiarów
Ogólną ideą jest stworzenie procedur składowanych, których moglibyśmy regularnie używać do wypełniania DWH -- tabel wymiarów oraz tabel faktów. Procedury te przeniosą dane między dwiema bazami danych na tym samym serwerze. Oznacza to, że niektóre zapytania wewnątrz tych procedur będą używać tabel z obu baz danych. Oczekuje się tego; musimy porównać stan DWH z bieżącą bazą danych i wprowadzić zmiany w DWH zgodnie z tym, co się dzieje w żywej bazie danych.
W naszym DWH mamy cztery tabele wymiarów:dim_time , dim_city , dim_product i dim_delivery_status .
Wymiar czasu jest wypełniany przez dodanie poprzedniej daty. Głównym założeniem jest to, że będziemy przeprowadzać tę procedurę codziennie, po zamknięciu działalności.
Wymiary miasta i produktu będą zależeć od bieżących wartości przechowywanych w city i product słowniki w aktywnej bazie danych. Jeśli dodamy coś do tych słowników, nowe wartości zostaną dodane do tabel wymiarów przy następnej aktualizacji DWH.
Ostatnia tabela wymiarów to dim_delivery_status stół. Nie zostanie zaktualizowany, ponieważ zawiera tylko trzy wartości domyślne. Dostawa jest w drodze, anulowana lub dostarczona.
Pomysł:Tabele faktów
Wypełnianie tabel faktów to właściwie prawdziwa praca. O ile słowniki w aktywnej bazie danych nie zawierają atrybutu timestamp, o tyle tabele z danymi wstawionymi w wyniku naszych operacji już tak. Zauważysz dwa atrybuty znacznika czasu, time_inserted i time_updated , w modelu danych.
Ponownie zakładam, że raz dziennie z powodzeniem uruchomimy import DWH. Dzięki temu możemy agregować dane na poziomie dziennym. Policzymy liczbę aktywnych i anulowanych klientów i subskrypcji, a także dostawy i dostarczone produkty w tym dniu.
Nasz model na żywo działa dobrze, jeśli uruchomimy procedurę wstawiania po COB (zamknięcie działalności). Jeśli jednak zależy nam na większej elastyczności, powinniśmy wprowadzić pewne zmiany w modelu. Jedną z takich zmian może być posiadanie oddzielnej tabeli historii do śledzenia dokładnego momentu, w którym zmieniły się dane związane z klientami lub subskrypcjami. W naszej obecnej organizacji będziemy wiedzieć, że nastąpiła zmiana, ale nie będziemy wiedzieć, czy przed tą zmianą były jakieś zmiany (np. klient zrezygnował wczoraj, ponownie aktywował swoje konto po północy, a następnie zrezygnował dzisiaj) .
Wypełnianie tabel wymiarów
Jak wspomniano wcześniej, przyjmę założenie, że będziemy uruchamiać import DWH dokładnie raz dziennie. Jeśli tak nie jest, potrzebowalibyśmy dodatkowego kodu, aby usunąć nowo wstawione dane z tabel wymiarów i faktów. W przypadku tabel wymiarów byłoby to ograniczone do usunięcia podanej daty.
Najpierw sprawdzimy, czy podana data istnieje w dim_time stół. Jeśli nie, dodamy nowy wiersz do tabeli; jeśli tak, nie musimy nic robić. W większości przypadków wszystkie daty są wstawiane podczas początkowego wdrożenia produkcyjnego. Ale pójdę z tym przykładem w celach edukacyjnych.
Dla dim_city i dim_product wymiary, dodam tylko nowe wartości wykryte w city i product tabele. Nie dokonam żadnych usunięć, ponieważ wszelkie wcześniej wstawione wartości mogą być przywoływane w jakiejś tabeli faktów. Moglibyśmy wybrać miękkie skreślenie, np. posiadanie flagi „aktywnej”, którą możemy włączać i wyłączać.
W przypadku ostatniej tabeli dim_delivery_status , nic nie zrobię, ponieważ zawsze będzie zawierał te same trzy wartości.
Poniższy kod tworzy procedurę, która zapełni tabele wymiarów dim_time i dim_city .
Dla wymiaru czasu dodam wczorajszą datę. Wychodzę z założenia, że proces ETL zaczyna się zaraz po północy. Sprawdzę, czy ten wymiar już istnieje, a jeśli nie, to dodam nową datę do tabeli.
W przypadku wymiaru miasta użyję LEFT JOIN, aby połączyć dane z aktywnej bazy danych i bazy danych DWH, aby określić, których wierszy brakuje. Następnie dodam tylko brakujące dane do tabeli wymiarów. Warto wspomnieć, że istnieje kilka sposobów sprawdzenia, czy dane uległy zmianie. Ten proces nazywa się przechwytywaniem zmian danych lub CDC. Powszechną metodą jest sprawdzanie zaktualizowanych sygnatur czasowych lub wersji. Istnieje kilka dodatkowych sposobów, ale wykraczają one poza zakres tego artykułu.
Przyjrzyjmy się teraz kodowi który jest napisany przy użyciu składni 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 ()
Uruchamianie tej procedury – co robimy za pomocą procedury z komentarzem CALL -- wstawia nową datę i wszystkie brakujące miasta do tabel wymiarów. Spróbuj dodać własny kod, aby wypełnić pozostałe dwie tabele wymiarów nowymi wartościami.
Proces ETL w hurtowni danych
Główną ideą hurtowni danych jest przechowywanie zagregowanych danych w pożądanym formacie. Oczywiście powinniśmy znać ten format jeszcze zanim zaczniemy budować magazyn. Jeśli zrobiliśmy wszystko zgodnie z planem, możemy uzyskać wszystkie korzyści, jakie oferuje nam DWH. Główną korzyścią jest lepsza wydajność podczas wykonywania zapytań. Nasze zapytania działają z mniejszą liczbą rekordów (ponieważ są agregowane) i działają w bazie danych raportowania (zamiast aktualnej).
Ale zanim będziemy mogli zapytać, musimy przechowywać fakty w naszej bazie danych. Sposób, w jaki to zrobimy, zależy od tego, co musimy później zrobić z naszymi danymi. Jeśli nie mamy dobrego ogólnego obrazu przed rozpoczęciem budowy naszego DWH, wkrótce możemy znaleźć się w tarapatach! wkrótce.
Nazwa tego procesu to ETL:E =Extract, T =Transform, L =Load. Przechwytuje dane, przekształca je w celu dopasowania do struktury DWH i ładuje je do DWH. Aby być precyzyjnym, faktycznym procesem, którego użyjemy, jest ELT:Extract, Load, Transform. Ponieważ korzystamy z procedur składowanych, wyodrębnimy dane, załadujemy je, a następnie przekształcimy zgodnie z naszymi potrzebami. Warto wiedzieć, że chociaż ETL i ELT są nieco inne, terminy są czasami używane zamiennie.
Wypełnianie tabel faktów
Wypełnianie tabel faktów jest powodem, dla którego naprawdę tu jesteśmy. Dzisiaj wypełnię dwie tabele faktów, fact_customer_subscribed tabela i fact_subscription_status stół. Pozostałe dwie tabele faktów możesz wypróbować jako pracę domową.
Zanim przejdziemy do wypełniania tabeli faktów, musimy założyć, że tabele wymiarów są wypełniane nowymi wartościami. Wypełnianie tabel faktów przebiega według tego samego wzorca. Ponieważ mają tę samą strukturę, wyjaśnię je razem.
Grupujemy dane według dwóch wymiarów:czasu i miasta. Wymiar czasu zostanie ustawiony na wczoraj, a identyfikator powiązanego rekordu znajdziemy w dim_time tabela porównując daty (ostatnie INNER JOIN w obu zapytaniach).
Identyfikator dim_city jest wyodrębniany przez połączenie wszystkich atrybutów, które tworzą UNIKALNĄ kombinację w tabeli wymiarów (nazwa miasta, kod pocztowy i nazwa kraju).
W tym zapytaniu przetestujemy wartości za pomocą CASE, a następnie zsumujemy je. Dla aktywnych i nieaktywnych klientów nie testowałem daty. Wybrałem jednak wartości takie, jakie są dla tych pól. W przypadku nowych i zlikwidowanych kont przetestowałem zaktualizowany czas.
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 ()
Po raz kolejny skomentowałem ostatnią linijkę. Usuń komentarz i możesz użyć tego wiersza, aby wywołać procedurę i wstawić nowe wartości. Pamiętaj, że nie usunąłem żadnych istniejących starych wartości, więc ta procedura nie zadziała, jeśli mamy już wartości dla tej daty i miasta. Można to rozwiązać, wykonując usuwanie przed wstawieniem.
Pamiętaj, że musimy wypełnić pozostałe tabele faktów w naszym DWH. Zachęcam do samodzielnego spróbowania!
Kolejną rzeczą, którą zdecydowanie polecam, to umieszczenie całego procesu w transakcji. To zapewniłoby, że albo wszystkie wstawienia się powiodą, albo nie zostaną wykonane żadne. Jest to bardzo ważne, gdy chcemy uniknąć częściowego wstawienia danych, np. jeśli mamy wiele procedur wstawiania wymiarów i faktów, a niektóre z nich wykonują swoją pracę, podczas gdy inne zawodzą.
Co myślisz?
Dziś widzieliśmy, jak możemy przeprowadzić proces ELT/ETL i załadować dane z działającej bazy danych do hurtowni danych. Chociaż proces, który zademonstrowaliśmy, jest dość uproszczony, zawiera wszystkie elementy potrzebne do E (wyodrębnienia) danych, T (przekształcenia) do odpowiedniego formatu, a na koniec L (oad) do DWH. Co myślisz? Podziel się z nami swoimi doświadczeniami w komentarzach poniżej.
