Neste artigo vamos abordar algumas alternativas para realizar uma carga incremental.
Basicamente, todos os exemplos seguem a mesma lógica que se baseia em comparar uma base de ORIGEM com uma base de DESTINO (DW) e:
1-Detectar novos registros. Registros que existem na origem mas não existem no destino
2-Detectar registros atualizados: Registros que existem em ambos bancos mas possui algum campo com valor diferente
3-Opcionalmente deletar registros no destino que foram deletados ou não existem mais na origem.
Para nossos exemplos, vamos criar algumas tabelas e inserir alguns dados:
--Criar bancos de origem e destino
CREATE DATABASE [SSISIncrementalLoad_Source]
CREATE DATABASE [SSISIncrementalLoad_Dest]
--Criar uma tabela no banco de origem
CREATE TABLE dbo.tblSource
(ColID int NOT NULL
,ColA varchar(10) NULL
,ColB datetime NULL constraint df_ColB default (getDate())
,ColC int NULL
,constraint PK_tblSource primary key clustered (ColID))
--Criar uma tabela no banco de destino
CREATE TABLE dbo.tblDest
(ColID int NOT NULL
,ColA varchar(10) NULL
,ColB datetime NULL
,ColC int NULL)
--Inserir alguns dados na origem
USE SSISIncrementalLoad_Source
GO
INSERT INTO dbo.tblSource (ColID,ColA,ColB,ColC) VALUES(0, 'A', '200180101', -1)
INSERT INTO dbo.tblSource (ColID,ColA,ColB,ColC) VALUES(1, 'B', '20180102', -2)
INSERT INTO dbo.tblSource (ColID,ColA,ColB,ColC) VALUES(2, 'N', '20180103', -3)
--Inserir alguns dados no destino
USE SSISIncrementalLoad_Dest
GO
--Inserir uma linha não alerada
INSERT INTO dbo.tblDest(ColID,ColA,ColB,ColC) VALUES(0, 'A', '200180101', -1)
--Inserir linha alterada
INSERT INTO dbo.tblDest (ColID,ColA,ColB,ColC) VALUES(1, 'C', '20190101', -2)
--Consultar novas linhas
SELECT s.ColID, s.ColA, s.ColB, s.ColC
FROM SSISIncrementalLoad_Source.dbo.tblSource s
LEFT JOIN SSISIncrementalLoad_Dest.dbo.tblDest d ON d.ColID = s.ColID
WHERE d.ColID IS NULL
Usando SQL Padrão para identificar linhas novas e alteradas:
Para iniciar, vamos aos exemplos básicos de como inserir, dados novos e atualizados:Para inserir novas linhas, o segredo está na cláusula WHERE onde são filtradas linhas que não existem no destino, logo são linhas novas.
--Inserir no destino apenas as linhas novas
INSERT INTO SSISIncrementalLoad_Dest.dbo.tblDest (ColID, ColA, ColB, ColC)
SELECT s.ColID, s.ColA, s.ColB, s.ColC
FROM SSISIncrementalLoad_Source.dbo.tblSource s
LEFT JOIN SSISIncrementalLoad_Dest.dbo.tblDest d ON d.ColID = s.ColID
WHERE d.ColID IS NULL
No processo de atualização, é feito um JOIN para identificar apenas as linhas que as colunas da origem e destino são diferentes e a atualização é feita apenas nestes casos.
--Atualizar dados no destino com as linhas alteradas.
UPDATE d
SET
d.ColA = s.ColA
,d.ColB = s.ColB
,d.ColC = s.ColC
FROM SSISIncrementalLoad_Dest.dbo.tblDest d
INNER JOIN SSISIncrementalLoad_Source.dbo.tblSource s ON s.ColID = d.ColID
WHERE (
(d.ColA != s.ColA)
OR (d.ColB != s.ColB)
OR (d.ColC != s.ColC)
)
Implementação do padrão utilizando o SSIS
Agora que vimos o padrão em sql, vamos criar um processo ETL utilizando o Integration Services.Crie um pacote e inclua um objeto control flow do tipo data flow. Utilize as tarefas de Lookup, Condicional split e Row Command para realizar a a tarefa.
1-Criar uma conexão Origem OLE DB: Conexão com tabela de origem
2-LEFT JOIN : Criar uma tarefa de Lookup apontando para a tabela de destino
Fazer a junção entre origem e destino através da coluna ColID e incluir o sulfixo Dest_ nas novas colunas retornadas. Isso irá ajudar na diferenciação.
Como por padrão a tarefa Lookup faz Inner Join e não Left Join, temos que marcar a opção de ignorar as falhas de saída. Isso vai fazer com que os casos que não houver correspondência exata entre os IDs de origem e destino, seja retornado NULL na coluna da tabela de destino. O valor NULL será usado para identificar a ausência do registro na tabela de destino.
3-Divisão Condicional: Criar uma tarefa "Condicional Split" para separar o fluxo de novas linhas e linhas atualizadas.
New Rows: Quando a coluna ID do destino for null, isso significa que o lookup não achou, logo é novo
Changed Rows: Compara as colunas da origem e destino para identificar as diferenças, se houver alguma retorna.
Unchanged: Todos os demais casos.
4-Destino OLE DB: Crie uma tarefa de destination para enviar e inserir o fluxo de novas linhas diretamente para a tabela de destino.
5-Update Destino Linha-por-linha: Inserir Linhas Alteradas com a tarefa OLE DB Command. Conecte o fluxo de linhas alteradas em uma tarega OLE DB Command.
Essa tarefa executa um comando sql para cada linha do fluxo de dados.
Configure as propriedades e informe o comando UPDATE com os parâmetros para serem mapeados na sequência.
Mapeando os Parâmetros
Fazer o mapeamento do fluxo de entrada de dados com o comando sql.
As colunas da tabela destino serão atualizadas com o conteúdo da tabela de origem com base no ID (mapeado na clausula WHERE).
A sequencia que os parâmetros aparecem no comando é a sequência de numeração do nome do parâmetro.
Esse padrão de carga incremental funciona bem, mas pode ser demorado para cargas muito volumosas, para isso existe uma variação para cargas maiores que realizam operações em conjuntos de dados e não em linha por linha, como no exemplo acima.
Padrão ETL Incremental Set-Based
Este exemplo utiliza uma tabela intermediaria stage para enviar as linhas alteradas e fazer a atualização em conjunto.
1- Truncate Stage Update: O primeiro passo é garantir que a tabela stage esteja truncada, sem linha alguma. Para isso utilize um control flow para executar um comando sql
2-Incremental Load Flow: O segundo passo, implementa o mesmo processo de carga padrão descrito anteriormente, mas ao invés de executar a atualização linha por linha, a tabela stage será populada com o fluxo de dados de linhas atualizadas.
3- Set-Based Update: A terceira etapa executa um comando SQL para realizar a atualização a tabela destino com os dados da stage
Pronto, desta maneira o processo de atualização é feito em conjunto, o processo desta forma é muito mais eficiente.
Carga Incremental utilizando o comando MERGE do T-SQL
Como alternativa ainda existe o comando MERGE disponível no T-SQL do SQL Server a partir de 2012.Ele aplica a mesma lógica em um código bem enxuto e de fácil leitura. Ele define a origem e destino, identifica a coluna de junção nos casos de linhas alteradas, especifica as colunas que serão comparadas e ainda traz a possibilidade de fazer a exclusão.
MERGE INTO [dbo].[tblDest] AS TGT --Alvo
USING [dbo].[tblSource] AS SRC --Fonte
--Faz a comparação das linhas de SRC com TGT
ON SRC.ColID = TGT.ColID
--Quando a linha na fonte for igual a linha que já existe no alvo e tiver algum campo com valor diferente entre elas
WHEN MATCHED AND ( TGT.ColA <> SRC.ColA OR TGT.ColB <> SRC.ColB OR TGT.ColC <> SRC.ColC)
THEN
UPDATE
SET TGT.ColA = SRC.ColA,
TGT.ColB = SRC.ColB,
TGT.ColC = SRC.ColC
--Quando a linha da fonte não existir no alvo, insere como nova linha
WHEN NOT MATCHED
THEN
INSERT
VALUES(SRC.ColID, SRC.ColA, SRC.ColB, SRC.ColC)
--Quando existe alguma linha no alvo que não existe na fonte. DELETE (opcional)
WHEN NOT MATCHED BY SOURCE THEN
DELETE;
Estas são algumas técnicas relacionadas ao processo de carga ETL incremental, muito útil em situações onde é necessário manter a integridade entre a origem e o destino mas não é possível realizar uma carga full devido ao volume de dados.
Fonte:
https://andyleonard.blog/2007/07/ssis-design-pattern-incremental-loads/
http://www.sqlservercentral.com/articles/Integration+Services+(SSIS)/72492/
https://docs.microsoft.com/pt-br/sql/t-sql/statements/merge-transact-sql?view=sql-server-2017
Nenhum comentário:
Postar um comentário