Исправление неточностей
This commit is contained in:
@@ -83,7 +83,7 @@ ALTER TABLE ods.products ADD PRIMARY KEY (product_id);
|
||||
|
||||
-- 4. DDS: интегрированная модель
|
||||
|
||||
-- dim_date: справочник дат (без первичного ключа — генерируется)
|
||||
-- dim_date: справочник дат (ключ — суррогатный date_key)
|
||||
CREATE TABLE dds.dim_date (
|
||||
date_key INT PRIMARY KEY,
|
||||
date_actual DATE NOT NULL,
|
||||
@@ -118,9 +118,9 @@ CREATE TABLE dds.dim_customer (
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
-- одна версия на момент времени
|
||||
-- одна версия на момент времени (на одну пару BK+valid_from)
|
||||
ALTER TABLE dds.dim_customer
|
||||
ADD CONSTRAINT uq_dim_customer_bk_from UNIQUE (customer_bk, valid_from); -- dim_product: измерение "
|
||||
ADD CONSTRAINT uq_dim_customer_bk_from UNIQUE (customer_bk, valid_from);
|
||||
|
||||
-- ускорители
|
||||
CREATE INDEX ix_dim_customer_bk_current ON dds.dim_customer (customer_bk) WHERE is_current;
|
||||
|
||||
@@ -128,56 +128,59 @@ SELECT product_id, name
|
||||
FROM ods.products;
|
||||
|
||||
-- 5. DDS: dim_customer — первичная загрузка SCD2 (full backfill из STG)
|
||||
-- Здесь мы пересчитываем всю историю клиента из событий в STG.
|
||||
-- В реальном DWH такую полную перезагрузку делают редко; для инкремента см. 03_demo_increment.sql и SCD.md.
|
||||
-- В ЭТОМ ДЕМО: dim_customer строится напрямую из stg.customers_raw, который играет роль
|
||||
-- устойчивого event-лога (все события по клиенту в одном месте).
|
||||
-- Это удобно для учебной первичной загрузки (full backfill), когда мы один раз
|
||||
-- восстанавливаем всю историю клиента.
|
||||
-- В РЕАЛЬНОМ DWH: так делают редко. Исторические измерения обычно строят
|
||||
-- поверх очищенных и нормализованных слоёв (ODS / PSA / Data Vault).
|
||||
-- Для примера инкрементальной заливки SCD2 по снимку из ODS см. 03_demo_increment.sql и SCD.md.
|
||||
TRUNCATE dds.dim_customer, dds.fact_sales;
|
||||
|
||||
BEGIN;
|
||||
WITH src AS (
|
||||
SELECT
|
||||
s.customer_id::INT AS customer_bk,
|
||||
NULLIF(trim(s.email), '') AS email,
|
||||
NULLIF(trim(s.phone), '') AS phone,
|
||||
NULLIF(trim(s.city), '') AS city,
|
||||
COALESCE(NULLIF(s.event_ts,'')::timestamp, s._load_ts,
|
||||
to_timestamp(regexp_replace(s._load_id,'^batch_',''),'YYYYMMDD_HH24MI')) AS eff_ts,
|
||||
dds.customer_hash(s.email, s.phone, s.city) AS hashdiff
|
||||
FROM stg.customers_raw s
|
||||
WHERE s.customer_id ~ '^\d+$'
|
||||
),
|
||||
ordered AS (
|
||||
SELECT *,
|
||||
lag(hashdiff) OVER (PARTITION BY customer_bk ORDER BY eff_ts) AS prev_hash,
|
||||
row_number() OVER (PARTITION BY customer_bk ORDER BY eff_ts) AS rn
|
||||
FROM src
|
||||
),
|
||||
changes AS (
|
||||
-- только первые состояния и фактические изменения атрибутов
|
||||
SELECT *
|
||||
FROM ordered
|
||||
WHERE prev_hash IS DISTINCT FROM hashdiff OR prev_hash IS NULL
|
||||
),
|
||||
framed AS (
|
||||
SELECT
|
||||
customer_bk, email, phone, city, hashdiff,
|
||||
CASE WHEN rn = 1 THEN timestamp '1900-01-01' ELSE eff_ts END AS valid_from, -- первая версия: техническое "начало истории"
|
||||
lead(eff_ts) OVER (PARTITION BY customer_bk ORDER BY eff_ts) AS next_ts
|
||||
FROM changes
|
||||
)
|
||||
INSERT INTO dds.dim_customer (
|
||||
customer_bk, email, phone, city, hashdiff,
|
||||
valid_from, valid_to, is_current,
|
||||
created_at, updated_at
|
||||
)
|
||||
WITH src AS (
|
||||
SELECT
|
||||
customer_bk, email, phone, city, hashdiff,
|
||||
valid_from,
|
||||
COALESCE(next_ts - interval '1 second', timestamp '9999-12-31') AS valid_to,
|
||||
(next_ts IS NULL) AS is_current,
|
||||
now(), now()
|
||||
FROM framed
|
||||
ORDER BY customer_bk, valid_from;
|
||||
COMMIT;
|
||||
s.customer_id::INT AS customer_bk,
|
||||
NULLIF(trim(s.email), '') AS email,
|
||||
NULLIF(trim(s.phone), '') AS phone,
|
||||
NULLIF(trim(s.city), '') AS city,
|
||||
COALESCE(NULLIF(s.event_ts,'')::timestamp, s._load_ts,
|
||||
to_timestamp(regexp_replace(s._load_id,'^batch_',''),'YYYYMMDD_HH24MI')) AS eff_ts,
|
||||
dds.customer_hash(s.email, s.phone, s.city) AS hashdiff
|
||||
FROM stg.customers_raw s
|
||||
WHERE s.customer_id ~ '^\d+$'
|
||||
),
|
||||
ordered AS (
|
||||
SELECT *,
|
||||
lag(hashdiff) OVER (PARTITION BY customer_bk ORDER BY eff_ts) AS prev_hash,
|
||||
row_number() OVER (PARTITION BY customer_bk ORDER BY eff_ts) AS rn
|
||||
FROM src
|
||||
),
|
||||
changes AS (
|
||||
-- только первые состояния и фактические изменения атрибутов
|
||||
SELECT *
|
||||
FROM ordered
|
||||
WHERE prev_hash IS DISTINCT FROM hashdiff OR prev_hash IS NULL
|
||||
),
|
||||
framed AS (
|
||||
SELECT
|
||||
customer_bk, email, phone, city, hashdiff,
|
||||
CASE WHEN rn = 1 THEN timestamp '1900-01-01' ELSE eff_ts END AS valid_from, -- первая версия: техническое "начало истории"
|
||||
lead(eff_ts) OVER (PARTITION BY customer_bk ORDER BY eff_ts) AS next_ts
|
||||
FROM changes
|
||||
)
|
||||
INSERT INTO dds.dim_customer (
|
||||
customer_bk, email, phone, city, hashdiff,
|
||||
valid_from, valid_to, is_current,
|
||||
created_at, updated_at
|
||||
)
|
||||
SELECT
|
||||
customer_bk, email, phone, city, hashdiff,
|
||||
valid_from,
|
||||
COALESCE(next_ts - interval '1 second', timestamp '9999-12-31') AS valid_to,
|
||||
(next_ts IS NULL) AS is_current,
|
||||
now(), now()
|
||||
FROM framed
|
||||
ORDER BY customer_bk, valid_from;
|
||||
|
||||
-- 6. DDS: fact_sales — загрузка фактов с учётом SCD
|
||||
-- В продакшене — фильтруем по диапазону дат (инкрементально)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
-- ===============================================
|
||||
-- Проверки качества данных после загрузки STG→ODS→DDS
|
||||
-- Запускается после 02_dml.sql
|
||||
-- Запускается после 02_dml_stg-dds.sql (и, при необходимости, 03_demo_increment.sql)
|
||||
-- ===============================================
|
||||
|
||||
-- 1. Проверка: dim_customer не пуста
|
||||
@@ -40,4 +40,3 @@ BEGIN
|
||||
FORMAT('ОШИБКА: у клиента 101 только %s версия, ожидается ≥2 (должна быть история)', version_count);
|
||||
RAISE NOTICE '✅ SCD Type 2: клиент 101 имеет %s версий — история сохранена', version_count;
|
||||
END $$;
|
||||
|
||||
|
||||
@@ -34,7 +34,6 @@ GROUP BY d.date_actual, p.product_name,
|
||||
-- Здесь — НЕ используем is_current! Нам нужна вся история для расчёта LTV
|
||||
-- Для кого: CRM-менеджер, retention-спец
|
||||
-- Пример использования: «Найти клиентов с LTV > 250 ₽ и email из Москвы для email-рассылки»
|
||||
«Найти клиентов с LTV > 250 ₽ и email из Москвы для email-рассылки»
|
||||
INSERT INTO dm.mart_customer_360 (
|
||||
customer_bk, first_order_date, last_order_date,
|
||||
total_orders, total_items, lifetime_value,
|
||||
@@ -44,7 +43,7 @@ SELECT
|
||||
c.customer_bk,
|
||||
MIN(d.date_actual) AS first_order_date,
|
||||
MAX(d.date_actual) AS last_order_date,
|
||||
COUNT(DISTINCT f.sale_id) AS total_orders,
|
||||
COUNT(DISTINCT f.sale_id) AS total_orders, -- считаем строки факта (продажи), не бизнес-заказы
|
||||
SUM(f.quantity) AS total_items,
|
||||
SUM(f.amount) AS lifetime_value,
|
||||
-- Берём email и город из самой свежей версии клиента
|
||||
|
||||
@@ -8,7 +8,8 @@
|
||||
-- 1) Загрузите CSV в stg.customer_status_raw (через COPY или \copy в psql).
|
||||
-- См. пример структуры файла в dwh-modeling/data/customer_status_events.csv
|
||||
-- 2) Переложите данные в ods.customer_status с приведением типов.
|
||||
-- customer_id → INT, status → VARCHAR(20), event_ts / load_ts → TIMESTAMP.
|
||||
-- customer_id → INT, status → VARCHAR(20), event_ts / load_ts → TIMESTAMP
|
||||
-- (в STG/ODS эта колонка будет жить как _load_ts).
|
||||
-- 3) Постройте из ods.customer_status измерение dds.dim_customer_status в стиле SCD2:
|
||||
-- - одна строка на период действия статуса (valid_from / valid_to);
|
||||
-- - is_current = TRUE только у актуальной строки для клиента;
|
||||
@@ -45,4 +46,3 @@
|
||||
-- );
|
||||
-- Идея: на каждую дату взять актуальный статус клиента
|
||||
-- через JOIN dds.dim_customer_status + dds.dim_date.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user