Nas duas partes anteriores, apresentamos o modelo de banco de dados ativo para um negócio baseado em assinatura e um data warehouse (DWH) que poderíamos usar para relatórios. Embora seja óbvio que eles devem trabalhar juntos, não havia conexão entre esses dois modelos. Hoje, daremos o próximo passo e escreveremos o código para transferir dados do banco de dados ativo para nosso DWH.
Os modelos de dados
Antes de mergulharmos no código, vamos nos lembrar dos dois modelos com os quais trabalharemos. O primeiro é o modelo de dados transacionais que usaremos para armazenar nossos dados em tempo real. Levando em consideração que administramos um negócio baseado em assinatura, precisaremos armazenar detalhes de clientes e assinaturas, pedidos de clientes e status de pedidos.
Há realmente muito que podemos adicionar a esse modelo, como rastrear pagamentos e armazenar dados históricos (especialmente alterações nos dados de clientes e assinaturas). Para enfatizar o processo ETL (extrair, transformar e carregar), porém, quero manter esse modelo o mais simples possível.
Usar um modelo de dados transacional como banco de dados de relatórios pode funcionar em alguns casos, mas não em todos os casos. Já mencionamos isso, mas vale a pena repetir. Se quisermos separar nossas tarefas de relatórios de nossos processos em tempo real, devemos criar algum tipo de banco de dados de relatórios. Um data warehouse é uma solução.
Nosso DWH está centrado em quatro tabelas de fatos. Os dois primeiros rastreiam o número de clientes e assinaturas em nível diário. Os dois restantes rastreiam o número de entregas e os produtos incluídos nessas entregas.
Minha suposição é que executaremos nosso processo ETL uma vez por dia. Primeiro, preencheremos as tabelas de dimensão com novos valores (quando necessário). Depois disso, preencheremos as tabelas de fatos.
Para evitar repetições desnecessárias, demonstrarei apenas o código que preencherá as duas primeiras tabelas de dimensão e as duas primeiras tabelas de fatos. As tabelas restantes podem ser preenchidas usando um código muito semelhante. Eu encorajo você a escrever o código você mesmo. Não há melhor maneira de aprender algo novo do que tentando.
A ideia:tabelas de dimensões
A ideia geral é criar procedimentos armazenados que possamos usar regularmente para preencher o DWH -- tabelas de dimensões, bem como tabelas de fatos. Esses procedimentos transferirão dados entre dois bancos de dados no mesmo servidor. Isso significa que algumas consultas dentro desses procedimentos usarão tabelas de ambos os bancos de dados. Isso é esperado; precisamos comparar o estado do DWH com o banco de dados ativo e fazer alterações no DWH de acordo com o que está acontecendo no banco de dados ativo.
Temos quatro tabelas de dimensão em nosso DWH:
dim_time , dim_city , dim_product e dim_delivery_status . A dimensão de tempo é preenchida adicionando a data anterior. A principal premissa é que executaremos esse procedimento diariamente, após o fechamento dos negócios.
A cidade e as dimensões do produto dependerão dos valores atuais armazenados na
city e product dicionários no banco de dados ao vivo. Se adicionarmos algo a esses dicionários, novos valores serão adicionados às tabelas de dimensão na próxima atualização do DWH. A última tabela de dimensão é o
dim_delivery_status tabela. Ele não será atualizado porque contém apenas três valores padrão. Uma entrega está em trânsito, cancelada ou entregue. A ideia:tabelas de fatos
Preencher tabelas de fatos é realmente o trabalho real. Embora os dicionários no banco de dados ativo não contenham um atributo timestamp, as tabelas com dados inseridos como resultado de nossas operações contêm. Você notará dois atributos de carimbo de data/hora,
time_inserted e time_updated , no modelo de dados. Novamente, estou assumindo que executaremos com sucesso a importação de DWH uma vez por dia. Isso nos permite agregar os dados em um nível diário. Contaremos o número de clientes e assinaturas ativos e cancelados, bem como as entregas e produtos entregues para essa data.
Nosso modelo ao vivo funciona bem se executarmos um procedimento de inserção após o COB (fechamento de negócios). Ainda assim, se quisermos mais flexibilidade, devemos fazer algumas mudanças no modelo. Uma dessas alterações pode ser ter uma tabela de histórico separada para rastrear o momento exato em que quaisquer dados relacionados a clientes ou assinaturas foram alterados. Com nossa organização atual, saberemos que a alteração aconteceu, mas não saberemos se houve alguma alteração antes desta (por exemplo, um cliente cancelou ontem, reativou sua conta após a meia-noite e cancelou novamente hoje) .
Preenchendo tabelas de dimensões
Como mencionado anteriormente, vou assumir que executaremos a importação DWH exatamente uma vez por dia. Se não for esse o caso, precisaríamos de código adicional para excluir dados recém-inseridos das tabelas de dimensões e fatos. Para as tabelas de dimensão, isso se limitaria à exclusão da data especificada.
Primeiro, verificaremos se a data especificada existe no
dim_time tabela. Caso contrário, adicionaremos uma nova linha à tabela; se isso acontecer, não precisamos fazer nada. Na maioria dos casos, todas as datas são inseridas durante a implantação de produção inicial. Mas vou usar este exemplo para fins educacionais. Para a
dim_city e dim_product dimensões, adicionarei apenas os novos valores que detectar na city e product mesas. Não farei nenhuma exclusão porque quaisquer valores inseridos anteriormente podem ser referenciados em alguma tabela de fatos. Poderíamos ir com uma exclusão suave, por exemplo ter um sinalizador "ativo" que poderíamos ligar e desligar. Para a última tabela,
dim_delivery_status , não farei nada porque sempre conterá os mesmos três valores. O código abaixo cria um procedimento que preencherá as tabelas de dimensão
dim_time e dim_city . Para a dimensão de tempo, adicionarei a data de ontem. Estou partindo do pressuposto de que o processo de ETL começa logo após a meia-noite. Vou verificar se essa dimensão já existe e, caso não exista, adiciono a nova data na tabela.
Para a dimensão da cidade, usarei um LEFT JOIN para juntar dados do banco de dados ao vivo e do banco de dados DWH para determinar quais linhas estão faltando. Em seguida, adicionarei apenas os dados ausentes à tabela de dimensões. Vale ressaltar que existem algumas maneiras de verificar se os dados foram alterados. Esse processo é chamado de captura de dados de alteração ou CDC. Um método comum é verificar se há registros de data e hora ou versões atualizadas. Existem algumas maneiras adicionais, mas elas estão fora do escopo deste artigo.
Vamos dar uma olhada no código agora, que é escrito usando a sintaxe do 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 ()
Executando este procedimento -- que fazemos usando o procedimento comentado CALL -- insere uma nova data e todas as cidades ausentes nas tabelas de dimensão. Tente adicionar seu próprio código para preencher as duas tabelas de dimensão restantes com novos valores.
O Processo ETL em um Data Warehouse
A principal ideia por trás do armazenamento de dados é conter dados agregados no formato desejado. Claro, devemos conhecer esse formato antes mesmo de começar a construir o armazém. Se fizermos tudo conforme o planejado, podemos obter todos os benefícios que um DWH nos oferece. O principal benefício é o desempenho aprimorado ao executar consultas. Nossas consultas funcionam com menos registros (porque são agregadas) e são executadas no banco de dados de relatórios (e não no banco de dados ativo).
Mas antes que possamos consultar, precisamos armazenar fatos em nosso banco de dados. A maneira como faremos isso depende do que precisamos fazer com nossos dados posteriormente. Se não tivermos uma boa visão geral antes de começar a construir nosso DWH, em breve poderemos nos encontrar em apuros! em breve.
O nome desse processo é ETL:E =Extrair, T =Transformar, L =Carregar. Ele pega os dados, os transforma para se adequar à estrutura DWH e os carrega no DWH. Para ser preciso, o processo real que usaremos é ELT:Extract, Load, Transform. Como estamos usando procedimentos armazenados, vamos extrair dados, carregá-los e transformá-los para atender às nossas necessidades. É bom saber que, embora ETL e ELT sejam um pouco diferentes, os termos às vezes são usados de forma intercambiável.
Preenchendo as tabelas de fatos
O preenchimento de tabelas de fatos é o motivo pelo qual estamos realmente aqui. Hoje, preencherei duas tabelas de fatos, a
fact_customer_subscribed tabela e o fact_subscription_status tabela. As duas tabelas de fatos restantes são suas para tentar como lição de casa. Antes de passarmos para o preenchimento da tabela de fatos, devemos assumir que as tabelas de dimensão são preenchidas com novos valores. O preenchimento das tabelas de fatos segue o mesmo padrão. Como eles têm a mesma estrutura, vou explicar os dois juntos.
Estamos agrupando os dados por duas dimensões:tempo e cidade. A dimensão de tempo será definida como ontem, e encontraremos o ID do registro relacionado no
dim_time tabela comparando datas (o último INNER JOIN em ambas as consultas). O ID de
dim_city é extraído juntando todos os atributos que formam uma combinação ÚNICA na tabela de dimensões (nome da cidade, código postal e nome do país). Nesta consulta, testaremos valores com CASE e, em seguida, SOMOS eles. Para clientes ativos e inativos, não testei a data. No entanto, selecionei valores como estão para esses campos. Para contas novas e canceladas, testei o horário atualizado.
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 ()
Mais uma vez, comentei a última linha. Remova o comentário e você pode usar esta linha para chamar o procedimento e inserir novos valores. Observe que não excluí nenhum valor antigo existente, portanto, esse procedimento não funcionará se já tivermos valores para essa data e cidade. Isso pode ser resolvido executando exclusões antes das inserções.
Lembre-se, precisamos preencher as tabelas de fatos restantes em nosso DWH. Eu encorajo você a tentar isso você mesmo!
Outra coisa que eu recomendaria definitivamente é colocar todo o processo dentro de uma transação. Isso garantiria que todas as inserções fossem bem-sucedidas ou nenhuma fosse feita. Isso é muito importante quando queremos evitar que os dados sejam inseridos parcialmente, por exemplo. se tivermos vários procedimentos para inserir dimensões e fatos e alguns deles fizerem seu trabalho enquanto outros falharem.
O que você acha?
Hoje vimos como poderíamos realizar o processo ELT/ETL e carregar dados de um banco de dados ativo em um data warehouse. Embora o processo que demonstramos seja bastante simplificado, ele contém todos os elementos necessários para E(extrair) os dados, T(transformar) em um formato adequado e, finalmente, L(oad) para o DWH. O que você acha? Por favor, conte-nos suas experiências nos comentários abaixo.
