Real Time BI e Change Data Capture

Real-time business intelligence (RTBI) é o processo de fornecer informações sobre as operações de negócios assim que elas ocorrem.

As operações de negócio assim que ocorrem são alimentadas no sistema RTBI em tempo real (Ou Near real time) que demonstra o estado atual da empresa; permitindo assim não somente as clássicas visões do passado do sistema de DW. Fornecendo suporte tático em tempo real para impulsionar ações de empresas para que reajam imediatamente aos eventos que ocorrerem.

Uma das tecnologias que melhor fornece a habilidade de carga de dados Real Time e o CDC (Change Data Capture).

CDC é um conjunto de padrões de software usados ​​para determinar (e rastrear) os dados que foram alterados de modo que uma ação pode ser tomada utilizando os dados alterados. Além disso, Change Data Capture (CDC) é uma abordagem para a integração de dados que se baseia na identificação, captura e entrega das modificações realizadas nas fontes de dados corporativas.

Exemplo de como ativar o CDC no SQL Server 2008:

1- O Serviço Agent do SQL Server 2008 precisa estar ativo.

2- Código para habilitar o CDC.

/*Habilita o Banco para uso do CDC*/

USE DB_NAME

GO

exec sys.sp_cdc_enable_db

-- Mostra os BDs com CDC ativo

select name, is_cdc_enabled from sys.databases

/*Habilita a tabela para o CDC*/

exec sys.sp_cdc_enable_table

    @source_schema = 'dbo',

    @source_name = 'TABLE_NAME' ,

    @role_name = 'CDCRole',

    @supports_net_changes = 1

/*Esse exemplo habilita as colunas a serem rastreadas*/

EXEC sys.sp_cdc_enable_table

@source_schema = N'dbo',

@source_name   = N'TABLE_NAME',

@role_name     = NULL,

@captured_column_list = '[COLUMN_NAME1],[ COLUMN_NAME2]'

   

/*Lista Tabelas rastreadas pelo CDC*/

select name, type, type_desc, is_tracked_by_cdc from sys.tables

where is_tracked_by_cdc = 1

-- ou

EXEC sys.sp_cdc_help_change_data_capture

Real Time BI and Change Data Capture (English Version)

Real-time business intelligence (RTBI) is the process of delivering information about business operations as they occur.

Business operations as they occur are fed into the RTBI system in real time (Or Near real time) which shows the current state of the company; allowing not only the classical visions of the past of classical DW system. Providing real-time tactical support to boost shares of companies to immediately respond to events as they occur.

One of the technologies that best provides the ability to load data and Real Time CDC (Change Data Capture).

CDC is a set of software design patterns used to determine (and track) the data that has changed so that action can be taken using the changed data. Also, Change data capture (CDC) is an approach to data integration that is based on the identification, capture and delivery of the changes made to enterprise data sources.

Example of how to enable CDC in SQL Server 2008:

1 - The SQL Server Agent service needs to be active.

2 - Code to enable the CDC.

/* Enables the database to use the CDC*/

USE DB_NAME

GO

exec sys.sp_cdc_enable_db

-- Shows DBs with active CDC

select name, is_cdc_enabled from sys.databases

/* Enables the table for use CDC */

exec sys.sp_cdc_enable_table

    @source_schema = 'dbo',

    @source_name = 'TABLE_NAME' ,

    @role_name = 'CDCRole',

    @supports_net_changes = 1

/* This example enables the columns to be tracked*/

EXEC sys.sp_cdc_enable_table

@source_schema = N'dbo',

@source_name   = N'TABLE_NAME',

@role_name     = NULL,

@captured_column_list = '[COLUMN_NAME1],[ COLUMN_NAME2]'

   

/* Tables listed to be tracked by CDC*/

select name, type, type_desc, is_tracked_by_cdc from sys.tables

where is_tracked_by_cdc = 1

-- or

EXEC sys.sp_cdc_help_change_data_capture

Surrogate Key (English Version)

Surrogate Key

The Surrogate word can be interpreted as an artificial synthetic product that is used to replace a natural product; Surrogate = Substitute; Applying this definition to dimensional modeling can say that a synthetic key is an artificial key that is used to replace a natural key the transactional environment.

The Surrogate Key (numeric, incremental and meaningless) arises from a concern not to have an informational database, linked and dependent on a transactional database. That is, has the benefit of immunize the DW  of the environmental changes and operational changes. Having the key independent of all other columns insulates possible changes in relationships of values ​​from the database or database design (making the database faster) ensuring the singularities.

Usually generate the surrogate key column based on a numeric data type which varies storing bytes according to the value stored. That is, if a field NUMERIC (10) is used in the definition of a Surrogate Key value will not be stored ten positions indicated in the definition of the field, but the actual value stored with the limit of ten positions. In Oracle this characteristic can be observed using the function VSIZE.

ETL tools have several tasks for the generation of Surrogates Key's as SAS Data Integration Studio, SAP Data Services, IBM Data Stage and etc..

DBMS solutions can also provide the generation of these keys as the data type  IDENTITY of SQL Server and  Oracle SEQUENCE function.

Why use Surrogate Key:

•        Keys transactional can be reused;

•        Treatment of historic dimensions;

•        Saving space;

•        Enable the integration of databases that have not integrated codes matching in transactional databases;

•        Enable the creation of control values ​​without including in transactional environment. Example: Client not registered, open date, etc..;

•        Improved performance in joins between tables.

How should be a Surrogate Key:

•        Must be an integer;

•        Sequential numbering increasing;

•        Must start at 1 until the required number;

•        There must have implicit meaning;

•        Shouldn't be a composite of transactional key.

Surrogate Key

Surrogate Key

A palavra Surrogate pode ser vista como um produto artificial sintético que é utilizado para substituir um produto natural; Surrogate = Substituta; Aplicando essa definição à modelagem dimensional pode-se dizer que uma chave sintética é uma chave artificial que é utilizada para substituir uma chave natural do ambiente transacional.

A Surrogate Key (campo numérico, incremental e sem significado) nasce da preocupação de não se ter uma base de dados informacional, vinculada e dependente de uma base transacional. Ou seja, possui como benefício imunizar o DW de mudanças de ambiente e de alterações operacionais. Ter a chave independente de todas as outras colunas isola possíveis mudanças de valores nos relacionamentos do banco de dados ou projeto do banco de dados (tornando a base de dados mais ágil) garantindo as singularidades.

Costumamos gerar a Surrogate Key baseada em uma coluna do tipo de dados numérico o qual varia o armazenamento de bytes de acordo com o valor armazenado. Ou seja, se um campo NUMERIC(10) for utilizado na definição de uma Surrogate Key o valor armazenado não será das dez posições indicadas na definição do campo, mas sim a do valor real armazenado com o limite de dez posições. No Oracle essa característica pode ser observada utilizando-se a função VSIZE.

Várias ferramentas de ETL possuem tarefas para a geração de Surrogates Key's , como SAS Data Integration Studio, SAP Data Services, IBM Data Stage e etc.

Soluções de BD's também podem prover a geração destas chaves como o tipo de Dados IDENTITY do SQL Server e a função SEQUENCE do Oracle.

Porque utilizar as Surrogate Key's:

·         Chaves transacionais podem ser reutilizadas;

·         Tratamento de histórico nas dimensões;

·         Economia de espaço;

·         Possibilitar a integração de bases de dados não integradas que possuam códigos transacionais coincidentes;

·         Possibilitar a criação de valores de “controle” sem implicar na sua inclusão no ambiente transacional. Exemplo: Cliente não cadastrado, Data em aberto, etc.;

·         Melhoria de performance nos joins entre tabelas.

Como deve ser uma Surrogate Key:

·         Deve ser um número inteiro;

·         Numeração seqüencial crescente;

·         Deve iniciar em 1 até o número necessário;

·         Não deve ter Significado implícito;

·         Não deve ser uma composição de chaves do transacional.

Particionamento Dinâmico - Dynamic Partitioning

A Stored Procedure (SQL Server) a seguir realiza o particionamento de uma tabela de nome "FACT_TABLE", o particionamento também tem a função de realizar a manutenção de expurgo da tabela. O código abaixo armazena os últimos 30 dias carregados na tabela "FACT_TABLE", e mantém o histórico dos meses fechados armazenando somente o último dia de cada mês já fechado.
Muito útil para processos que acompanham vendas diárias, mas precisam manter a última posição fechada; posições de Carteiras e etc.

The Stored Procedure (SQL Server) performs the partitioning of a table named "FACT_TABLE" this partitioning also has the function to perform the purge o the table. The code below stores the last 30 days in the loaded table "FACT_TABLE" and keeps track of the months closed storing only the last day of each month closed.
Very useful for processes like daily sales, but need to keep the last closed position; positions and etc.


Código abaixo (Code below)

--Cria função de partição da tabela de fato. (Create the partition function of the fact table)
------------------------------------------------------------------------------------------------------------
DECLARE @DATA SMALLDATETIME

SET @DATA = '1900-01-01'


CREATE PARTITION FUNCTION PFDataRef (SMALLDATETIME)
AS RANGE RIGHT FOR VALUES (@DATA)

GO

-- Cria função de partição da tabela auxiliar. (Creates the partition function of the auxiliary table)
------------------------------------------------------------------------------------------------------------
DECLARE @DATA SMALLDATETIME

SET @DATA = '1900-01-01'

CREATE PARTITION FUNCTION PFDataRefAUX (SMALLDATETIME)
AS RANGE RIGHT FOR VALUES (@DATA)

GO

-- Cria a partição da tabela de fato. (Create the partition of the fact table)
------------------------------------------------------------------------------------------------------------
CREATE PARTITION SCHEME PSDataRef1
AS PARTITION PFDataRef
ALL TO ([PRIMARY])

GO

-- Cria a partição da tabela dauxiliar. (Creates the partition of the auxiliary table)
------------------------------------------------------------------------------------------------------------
CREATE PARTITION SCHEME PSDataRefAUX1
AS PARTITION PFDataRefAUX
ALL TO ([PRIMARY])

GO

-- Cria a tabela de Fato. (Creates the Fact Table)
------------------------------------------------------------------------------------------------------------
CREATE TABLE FACT_TABLE( C1 SMALLDATETIME NOT NULL,
C2 INT NOT NULL) ON PSDataRefCNG1 (C1)

GO

-- Cria a tabela auxiliar. (Creates the Auxiliary Table)
------------------------------------------------------------------------------------------------------------
CREATE TABLE FACT_TABLE_AUX( C1 SMALLDATETIME NOT NULL,
C2 INT NOT NULL) ON PSDataRefCNGAUX1 (C1)

GO

-- Stored Procedure para Particionamento. (Stored Procedure for Partitioning.)
------------------------------------------------------------------------------------------------------------
CREATE PROCEDURE SP_ParticaoExpurgo @DATA SMALLDATETIME
AS

DECLARE @DataPart SMALLDATETIME,
@DataSplit SMALLDATETIME,
@DataRange SMALLDATETIME,
@Cont INT


SET @DataRange =  cast((select top 1 [value] from sys.partition_range_values
where function_id = (select function_id
from sys.partition_functions
where name = 'PFDataRef')
order by boundary_id ASC) as smalldatetime)

IF @DataRange = @DATA
RETURN

ALTER PARTITION SCHEME PSDataRef1 NEXT USED [PRIMARY]
ALTER PARTITION FUNCTION PFDataRef() SPLIT RANGE (@DATA)

DECLARE Part_cursor CURSOR
FOR SELECT CAST( [value]  AS SMALLDATETIME) AS DATA
FROM sys.partition_range_values
WHERE function_id = (SELECT function_id
FROM sys.partition_functions
WHERE name = 'PFDataRef')
ORDER BY boundary_id DESC

OPEN Part_cursor

FETCH NEXT FROM Part_cursor INTO @DataPart

SET @Cont = 1

CREATE TABLE ##Part (DataPart SMALLDATETIME, Cont INT)

WHILE (@@fetch_status <> -1)
BEGIN

If @Cont <= 30
BEGIN
INSERT INTO ##Part VALUES (@DataPart, @Cont)
END

SET @Cont = @Cont + 1

FETCH NEXT FROM Part_cursor INTO @DataPart
END

CLOSE Part_cursor

DEALLOCATE Part_cursor

DECLARE Split_cursor CURSOR
FOR SELECT CAST( [value] AS SMALLDATETIME) AS DATA
FROM sys.partition_range_values
WHERE function_id = (SELECT function_id
FROM sys.partition_functions
WHERE name = 'PFDataRef')
AND CAST( [value] AS SMALLDATETIME) NOT IN (SELECT DataPart
FROM ##Part
WHERE Cont <= 30
UNION
SELECT MAX(CAST( [value]  AS SMALLDATETIME)) AS DATA
FROM sys.partition_range_values
WHERE function_id = ( select function_id
from sys.partition_functions
where name = 'PFDataRef')
GROUP BY YEAR(CAST( [value]  AS SMALLDATETIME)) * 100 + MONTH(CAST( [value]  AS SMALLDATETIME)) )
ORDER BY boundary_id ASC


OPEN Split_cursor

FETCH NEXT FROM Split_cursor INTO @DataSplit

WHILE (@@fetch_status <> -1)
BEGIN

ALTER PARTITION SCHEME PSDataRefAUX1 NEXT USED [PRIMARY]
ALTER PARTITION FUNCTION PFDataRefAUX() SPLIT RANGE (@DataSplit)
ALTER TABLE dbo.UFACT_TABLE SWITCH PARTITION $PARTITION.PFDataRef(@DataSplit) TO dbo.FACT_TABLE_AUX PARTITION $PARTITION.PFDataRefAUX(@DataSplit)
TRUNCATE TABLE dbo.FACT_TABLE_AUX
ALTER PARTITION FUNCTION PFDataRefAUX() MERGE RANGE (@DataSplit)
ALTER PARTITION FUNCTION PFDataRef() MERGE RANGE (@DataSplit)

FETCH NEXT FROM Split_cursor INTO @DataSplit
END

CLOSE Split_cursor

DEALLOCATE Split_cursor

DROP TABLE ##Part


-- Teste: (Test)
------------------------------------------------------------------------------------------------------------
exec SP_ParticaoExpurgo '2008-01-01'
INSERT INTO FACT_TABLE VALUES ( '2008-01-01' , 1 )

exec SP_ParticaoExpurgo '2008-01-02'
INSERT INTO FACT_TABLE VALUES ( '2008-01-02' , 2 )

exec SP_ParticaoExpurgo '2008-01-03'
INSERT INTO FACT_TABLE VALUES ( '2008-01-03' , 3 )

/* ... Continue até gerar a quantidade de registros necessários ao teste. Você também pode mudar o valor da procedure de 30 para outro para testar com um volume menor. */
/* ... Continue to generate the quantity of records needed to test. You can also change the value of the quantity of records (less then 30) to test with a smaller volume.*/

-- Consultas para apurar o funcionamento do particionamento. (Queries to test the partitioning operation.)
------------------------------------------------------------------------------------------------------------
SELECT * FROM FACT_TABLE

SELECT *
FROM sys.partition_range_values
WHERE function_id = (SELECT function_id
FROM sys.partition_functions
WHERE name = 'PFDataRef')
ORDER BY boundary_id DESC

SELECT * FROM FACT_TABLE_AUX

SELECT *
FROM sys.partition_range_values
WHERE function_id = (SELECT function_id
FROM sys.partition_functions
WHERE name = 'PFDataRefAux')
ORDER BY boundary_id DESC

SCD (Slowly Changing Dimensions) - English Version

SCD (Slowly Changing Dimensions)

Slowly Changing Dimensions means that data changes without a temporal basis. There isn't a planned change by time. There are various types of SCD (Type 0, 1, 2, 3, 4, 6-hybrid-). The most common are type 1, 2 and 3.

Below an example of each type:

Type I:

Suppose we have a Customer dimension with the following attributes: social security number, name and state, as shown below:

Where SK is the Surrogate Key (sequential, numerical and meaningless), the social security number is the business key, name a descriptive attribute and state an attribute that we are looking to update.

In SCD type I, when there is a change in a given data this value is updated. No version is generated.

In the case below, the customer Erick Furst changed from the state of MG for the state of RJ (Brazilian States).

  

Note that SK remains the same and we have only one instance of customer Erick Furst.

Type II:

The SCD type II allows versioning the registry.

In this scenario the SCD Type II makes use of two new fields, Start Date and End Date that will allow the recording of the client's situation (This technique can also be used with the field "Current Record Indicator" or the "Version Number" ).

Note that the client Erick Furst was added on September 24, 2012 (Start Date field) and End Date field with empty value or a date that means null (eg 12/31/1800 or 12/31/9999) indicates that this record is valid, ie is the current record.

Suppose the client Erick Furst moved to the RJ on September 25, 2012. The new version will be included with a new SK (Value 2). And the older version will receive an End Date equivalent to the entry of the new record and the newest version will have the end date open (Null).

Type III:

The SCD type III allows versioning by column.

Assume the same position above in the SCD type II, instead of the versioned record the new version will be included in columns. The disadvantage of this technique is that the versioning is limited to the number of columns provided for this purpose.

Eg:

SCD (Slowly Changing Dimensions)

Slowly Changing Dimensions (Dimensões de modificação lenta) significa que os dados mudam sem uma base temporal. Não é uma mudança prevista por tempo.

Existem vários tipos de SCD (Tipo 0, 1, 2, 3, 4, 6 -híbridos-). Os mais comuns -Por mais comuns entenda também que são utilizados em ferramentas ETL- são o tipo 1, 2 e 3.

Abaixo um exemplo de cada tipo:

Tipo I:

Suponha que tenhamos uma dimensão Cliente com os seguintes atributos: CPF, Nome e UF, conforme tabela abaixo:

Onde SK é a Surrogate Key (Chave numérica seqüencial e sem significado de negócio), o CPF é a chave de negócio, nome atributo descritivo e UF um atributo que se deseja atualizar.

No SCD tipo I, quando há a mudança de um dado este valor é atualizado. Ou seja não é gerada nenhuma versão.

No caso abaixo o cliente Erick Furst mudou do estado de MG para o estado do RJ.

Observe que a SK permanece a mesma e possuímos somente uma ocorrência do cliente Erick Furst.

Tipo II:

O SCD tipo II permite o versionamento do registro.

Nesse cenário o SCD Tipo II faz uso de dois novos campos Data Início Vigência e Data Fim Vigência que irão permitir o registro da situação do cliente (Essa técnica também pode ser utilizada com o campo “Indicador de Registro Corrente” ou com o “Número da Versão”).

Observe que o cliente Erick Furst foi adicionado no dia 24/09/2012 (campo Data Início Vigência) e o campo Data Fim Vigência vazio ou com uma data que representa nulo (ex.: 31/12/1800 ou 31/12/9999) indica que este registro está vigente, ou seja é o registro corrente.

Suponhamos que o cliente Erick Furst mudou-se para o RJ no dia 25/09/2012. A nova versão será incluída com uma nova SK (Valor 2). E a versão mais antiga receberá uma Data Fim Vigência equivalente à entrada do novo registro e a versão mais nova ficará com a Data fim Vigência em aberto.

Tipo III:

O SCD tipo III permite o versionamento por coluna.

Exemplo, suponha a mesma situação acima do SCD tipo II, ao invés de versionar o registro a nova versão será incluída em colunas. A desvantagem desta técnica é que o versionamento é limitado à quantidade de colunas previstas para tal.

Ex.:

SAS Macro para tarefa de Lookup

%let APPDIR=%SYSGET(FOLDER_NAME);
%include "&APPDIR\PROG\GECRPDeclaraLibNames.sas";
/* Atribuição de bibliotecas (Substitua o código acima com as suas bibliotecas) */

%macro GELookup(NomeBiblOut, TabOut, NomeBiblMan, TabMan, CmpChaveMan, CmpRetornoMan, NomeBiblPes, TabPes, CmpChavePes, CmpRetornoPes);
/*
    PROGRAMA SAS
         PROJ: Tarefas gerais de ETL.
         DESC: Realiza a tarefa de Lookup tendo uma tabela Mandatória e uma tabela de Pesquisa.
         AUTOR: Erick Furst erick.furst@coresynesis.com.br
         DATA: 2009-08-27
     INPUTS/OUTPUTS
        INs : Nome biblioteca e nome tabela de saída (output): NomeBiblOut, TabOut.
 Nome biblioteca, nome tabela de mandatória, campo(s) chave da tabela mandatória e campo(s)
de retorno da tabela mandatória: NomeBiblMan, TabMan, CmpChaveMan, CmpRetornoMan.
 Nome biblioteca, tabela de pesquisa, campo(s) chave da tabela pesquisa e campo(s) de retorno da
tabela pesquisa: NomeBiblPes, TabPes, CmpChavePes, CmpRetornoPes.
        OUTs: Tabela de saída (output).
    MODIFICACOES
        2009-09-08 erick.furst@coresynesis.com.br
             (+) Inclusão de chamada de Libs por variável de sistema.
*/

/* Recupera o tipo e o tamanho de dados dos campos da tabela mandatória. Já no formato para trabalho no DATA STEP.
E armazena na tabela WORK.TamMan*/
proc sql;
CREATE TABLE WORK.TamMan AS
select 'length ' || NAME || ' ' || CASE TYPE WHEN 'char' THEN '$' ELSE '' END || TRIM(PUT(LENGTH,6.)) || ';' AS TamCol
from dictionary.columns
where UPPER(libname)= UPPER(&NomeBiblMan)
and UPPER(MEMNAME) = UPPER(&TabMan);
run;

/* Recupera o tipo e o tamanho de dados dos campos da tabela de pesquisa. Já no formato para trabalho no DATA STEP.
E armazena na tabela WORK.TamPes*/
proc sql;
CREATE TABLE WORK.TamPes AS
select 'length ' || NAME || ' ' || CASE TYPE WHEN 'char' THEN '$' ELSE '' END || TRIM(PUT(LENGTH,6.)) || ';' AS TamCol
from dictionary.columns
where UPPER(libname)= UPPER(&NomeBiblPes)
and UPPER(MEMNAME) = UPPER(&TabPes);
run;

/* Desmembra a String armazenada na variável &CmpChaveMan (Campos chave da tabela mandatória)
para identificação dos campos. E concatenação posterior na string do call execute que gerará o código do lookup. */
data WORK.ChaveMan;

CmpChaveMan = &CmpChaveMan;
CmpChaveManTra = CmpChaveMan;
Pos = 1;
PosIni = 1;
PosFim = length(CmpChaveManTra);
OrdemMan = 0;

do while (PosIni < PosFim AND Pos ^= 0);

Pos = INDEX(CmpChaveManTra, ",");
CmpChaveManLst = SUBSTR(CmpChaveManTra, PosIni, Pos - 1);
CmpChaveManTra = SUBSTR(CmpChaveManTra, Pos + 1, PosFim - Pos);
OrdemMan = OrdemMan + 1;
output;

drop CmpChaveManTra; drop Pos; drop PosIni; drop PosFim;
end;

run;


/* Desmembra a String armazenada na variável &CmpChavePes (Campos chave da tabela de pesquisa)
para identificação dos campos. E concatenação posterior na string do call execute que gerará o código do lookup. */
data WORK.ChavePes;

CmpChavePes = &CmpChavePes;
CmpChavePesTra = CmpChavePes;
Pos = 1;
PosIni = 1;
PosFim = length(CmpChavePesTra);
OrdemPes = 0;

do while (PosIni < PosFim AND Pos ^= 0);

Pos = INDEX(CmpChavePesTra, ",");
CmpChavePesLst = SUBSTR(CmpChavePesTra, PosIni, Pos - 1);
CmpChavePesTra = SUBSTR(CmpChavePesTra, Pos + 1, PosFim - Pos);
OrdemPes = OrdemPes + 1;
output;

drop CmpChavePesTra; drop Pos; drop PosIni; drop PosFim;
end;

run;

/* DATA STEP que gera o código de lookup para a função CALL EXECUTE. */
data _null_;

/* Entradas do Lookup */
/* ****************** */
* Informar Tabela de saída;
NomeBiblOut = &NomeBiblOut;
TabOut = &TabOut;
* Informar Tabela de mandatória, campo(s) chave da tabela mandatória e campo(s) de retorno da tabela mandatória;
NomeBiblMan = &NomeBiblMan;
TabMan = &TabMan;
CmpChaveMan = &CmpChaveMan;
CmpRetornoMan = &CmpRetornoMan;
* Informar Tabela de pesquisa, campo(s) chave da tabela pesquisa e campo(s) de retorno da tabela pesquisa;
NomeBiblPes = &NomeBiblPes;
TabPes = &TabPes;
CmpChavePes = &CmpChavePes;
CmpRetornoPes = &CmpRetornoPes;


length TxtCodLookup $5000;
length TxtCodLookup2 $1000;
length TxtCodLookup3 $1000;
length TxtCodLookup4 $1000;


/* Tratamento de campos. */
/* ********************* */
/* Definição de padrão para substituição com a função "rxparse". Uso necessário para passagem de
campos de parâmetro separados por vírgula. A substituição de caracteres é feita através da função
"rxchange" e também é feita a limpeza de caracteres brancos através da função "compress". */

rx=rxparse("$',' to '"",""'");

* Chave da Tabela Mandatória;
CmpChaveManCon = &CmpChaveMan || '                                                                                                                                                                     ';
call rxchange(rx,length(CmpChaveManCon),CmpChaveManCon); /* Substituição de valor */
CmpChaveManCon=compress(CmpChaveManCon); /* Eliminação de espaços brancos. */

* Campos de retorno da Tabela Mandatória;
CmpRetornoManCon = &CmpRetornoMan || '                                                                                                                                                                     ';
call rxchange(rx,length(CmpRetornoManCon),CmpRetornoManCon); /* Substituição de valor */
CmpRetornoManCon=compress(CmpRetornoManCon); /* Eliminação de espaços brancos. */

* Chave da Tabela de Pesquisa;
CmpChavePesCon = &CmpChavePes || '                                                                                                                                                                     ';
call rxchange(rx,length(CmpChavePesCon),CmpChavePesCon); /* Substituição de valor */
CmpChavePesCon=compress(CmpChavePesCon); /* Eliminação de espaços brancos. */

* Campos de retorno da Tabela de Pesquisa;
CmpRetornoPesCon = CmpRetornoPes || '                                                                                                                                                                     ';
call rxchange(rx,length(CmpRetornoPesCon),CmpRetornoPesCon); /* Substituição de valor */
CmpRetornoPesCon=compress(CmpRetornoPesCon); /* Eliminação de espaços brancos. */


/* Monta String para execução do código de lookup na função CALL EXECUTE. */
TxtCodLookup = 'DATA ' || &NomeBiblOut || '.' || &TabOut || '; ';

/* O cursor abaixo monta o comando lenght para a tabela mandatória.*/
length TamCol $1000;
length TxtCodLkMan $5000;

declare hash LstTamColMan(dataset:"WORK.TamMan");

LstTamColMan.defineKey("TamCol");
LstTamColMan.defineData("TamCol");
LstTamColMan.defineDone();

declare hiter iterobj1('LstTamColMan');

crTamCol = iterobj1.first();

do while (crTamCol=0);

TxtCodLkMan = CATS(TxtCodLkMan, TamCol);

crTamCol = iterobj1.next();

end;

drop TamCol; drop crTamCol;

/* Concatenação dos blocos da string na variável final TxtCodLookup. */
TxtCodLookup = CATS(TxtCodLookup, ' ');
TxtCodLookup = CATS(TxtCodLookup, TxtCodLkMan);

TxtCodLookup2 =
' declare hash TabMandatoria(dataset:"' || &NomeBiblMan || '.' || &TabMan || '"); ' ||
'TabMandatoria.defineKey("' || TRIM(CmpChaveManCon) || '"); ' ||
'TabMandatoria.defineData("' || TRIM(CmpRetornoManCon) || '"); ' ||
'TabMandatoria.defineDone(); ';

TxtCodLookup = CATS(TxtCodLookup, TxtCodLookup2);


/* O cursor abaixo monta o comando lenght para a tabela de pesquisa.*/
length TamCol $1000;
length TxtCodLkPes $5000;

declare hash LstTamColPes(dataset:"WORK.TamPes");

LstTamColPes.defineKey("TamCol");
LstTamColPes.defineData("TamCol");
LstTamColPes.defineDone();

declare hiter iterobj2('LstTamColPes');

crTamCol = iterobj2.first();

do while (crTamCol=0);

TxtCodLkPes = CATS(TxtCodLkPes, TamCol);

crTamCol = iterobj2.next();

end;

drop TamCol; drop crTamCol;

/* Concatenação dos blocos da string na variável final TxtCodLookup. */
TxtCodLookup = CATS(TxtCodLookup, ' ');
TxtCodLookup = CATS(TxtCodLookup, TxtCodLkPes);

TxtCodLookup3 =
'declare hash TabPesquisa(dataset:"' || &NomeBiblPes || '.' || &TabPes || '"); ' ||
'TabPesquisa.defineKey("' || TRIM(CmpChavePesCon) || '"); ' ||
'TabPesquisa.defineData("' || TRIM(CmpRetornoPesCon) || '"); ' ||
'TabPesquisa.defineDone(); ' ||
'declare hiter iterobj(''TabMandatoria''); ' ||
'rc = iterobj.first(); ' ||
'do while (rc=0); ';

TxtCodLookup = CATS(TxtCodLookup, TxtCodLookup3);


/* O cursor abaixo monta o comando de comparação entre o(s) campo(s) chave da tabela de pesquisa com o(s)
campo(s) chave da tabela mandatória. Para a posterior recuperação de campos feito pelo comando FIND do objeto HASH.
Ex.: CampoChaveTabelaPesquisa = CampoChaveTabelaMandatoria; */
length OrdemMan 3; length CmpChaveManLst $32;

declare hash LstMan(dataset:"WORK.ChaveMan");

LstMan.defineKey("OrdemMan");
LstMan.defineData("OrdemMan","CmpChaveManLst");
LstMan.defineDone();

declare hiter iterobj3('LstMan');

length OrdemPes 3; length CmpChavePesLst $32;

declare hash LstPes(dataset:"WORK.ChavePes");

LstPes.defineKey("OrdemPes");
LstPes.defineData("OrdemPes","CmpChavePesLst");
LstPes.defineDone();

cursor = iterobj3.first();

do while (cursor=0);

OrdemPes = OrdemMan;
cursor = LstPes.find();
TxtCodLookup = CATS(TxtCodLookup, CmpChavePesLst || ' = ' || CmpChaveManLst || '; ');

cursor = iterobj3.next();

end;

drop CmpChavePesLst; drop CmpChaveManLst; drop OrdemPes; drop OrdemMan; drop cursor;

/* Concatenação dos blocos da string na variável final TxtCodLookup. */
TxtCodLookup4 =
' rc = TabPesquisa.find(); ' ||
' output; ' ||
' rc = iterobj.next(); ' ||
'end; ' ||
'drop rc; ' ||
'run; ';

TxtCodLookup = CATS(TxtCodLookup, TxtCodLookup4);

/* Gera e executa o código de lookup. */
CALL EXECUTE (TxtCodLookup);

drop _All_;

run;

%mend;


/* ----------------------------------------------------  */

/* Como usar */


%let APPDIR=%SYSGET(FOLDER_NAME);
%include "&APPDIR\PROG\GECRPDeclaraLibNames.sas";

/* Atribuição de bibliotecas (Substitua o código acima com as suas bibliotecas) */


%include "&APPDIR\PROG\GELookup.sas";
/* Arquivo com o código da macro %macro GELookup */

/* Entradas do Lookup, separar campos por vírgulas ( , ).*/
/* ***************************************************** */
* Informar Tabela de saída;
%let NomeBiblOut = 'LIBRARY_NAME';
%let TabOut = 'ResultadoLookup';  /* Nome da tabela de output */
/*-------------------------------------------------------*/
* Informar Tabela de mandatória, campo(s) chave da tabela mandatória e campo(s) de retorno da tabela mandatória;
%let NomeBiblMan = 'LIBRARY_NAME';
%let TabMan = 'MANDATORY_TABLE';
%let CmpChaveMan = 'KEY_COLUMN';
%let CmpRetornoMan = 'KEY_COLUMN';
/*-------------------------------------------------------*/
* Informar Tabela de pesquisa, campo(s) chave da tabela pesquisa e campo(s) de retorno da tabela pesquisa;
%let NomeBiblPes = 'LIBRARY_NAME';
%let TabPes = 'LOOKUP_TABLE';
%let CmpChavePes = 'KEY_COLUMN';
%let CmpRetornoPes = 'RETURNED_COLUMN_1,RETURNED_COLUMN_2';
/*-------------------------------------------------------*/

%GELookup(&NomeBiblOut, &TabOut, &NomeBiblMan, &TabMan, &CmpChaveMan, &CmpRetornoMan,
 &NomeBiblPes, &TabPes, &CmpChavePes, &CmpRetornoPes);

Macro SAS - SCD I e II

%let APPDIR=%SYSGET(NOME_DIRETORIO);

%include "&APPDIR\PROG\GECRPDeclaraLibNames.sas";

%macro GESCD(TipoSCD, TipoChv,
NomeBiblStg, TabStg, CmpChaveStg, CmpRetornoStg,
NomeBiblDim, TabDim, CmpChaveSkDim, CmpChaveDim,
NomeCmpDatIniVigDim, NomeCmpDatFimVigDim );
/*
    PROGRAMA SAS
         PROJ: Tarefas gerais de ETL.
         DESC: Realiza a tarefa de SCD tipos I e II  a partir de uma tabela de stage e uma dimensão.
         AUTOR: Erick Furst - erick.furst@coresynesis.com.br
         DATA: 2009-09-08
     INPUTS/OUTPUTS
        INs : Nome biblioteca, tabela, campo(s) chave de negócio e campo(s) de retorno da tabela
stage: NomeBiblStg, TabStg, CmpChaveStg, CmpRetornoStg.
 Nome biblioteca, tabela, campo(s) chave substituta, campo(s) chave de negócio e campo(s) de retorno da
tabela de dimensão: NomeBiblDim, TabDim, CmpChaveSkDim, CmpChaveDim.
 Obs.: (Só poderá ser informado um campo para a variável "CmpChaveSkDim".)
        OUTs: Não há.
    MODIFICACOES
        2009-09-10 - Erick Furst - erick.furst@coresynesis.com.br
             (+) Alteração para realização de SCD Tipo I sem surrogate Key.
        2009-09-12 - Erick Furst - erick.furst@coresynesis.com.br
             (+) Alteração para correção do campo de suurogate key que estava fixo qdo deveria utilizar a variável.
*/
/*-------------------------------------------------------------------------------------------------------------------*/

/* Armazena Data Atual do sistema para gravação do controle de histórico do SCD tipo 2. */
%let DataAtual = today( );

/* Recupera o valor da maior Sk armazenada na dimensão. */

%if &TipoChv = 'S' %then %do;
proc sql;
SELECT MAX( &CmpChaveSkDim ) into : Maior FROM &NomeBiblDim..&TabDim ;
quit;
%end;

/* Recupera o tipo e o tamanho de dados dos campos da tabela de Stage. Já no formato para trabalho no DATA STEP.
E armazena na tabela WORK.TamStg*/
proc sql;
CREATE TABLE WORK.TamStg AS
select 'length ' || NAME || ' ' || CASE TYPE WHEN 'char' THEN '$' ELSE '' END || TRIM(PUT(LENGTH,6.)) || ';' AS TamCol
from dictionary.columns
where UPPER(libname)= UPPER("&NomeBiblStg")
and UPPER(MEMNAME) = UPPER("&TabStg");
quit;

/* Recupera o tipo e o tamanho de dados dos campos da tabela de Dimensão. Já no formato para trabalho no DATA STEP.
E armazena na tabela WORK.TamDim*/
proc sql;
CREATE TABLE WORK.TamDim AS
select 'length ' || NAME || ' ' || CASE TYPE WHEN 'char' THEN '$' ELSE '' END || TRIM(PUT(LENGTH,6.)) || ';' AS TamCol
from dictionary.columns
where UPPER(libname)= UPPER("&NomeBiblDim")
and UPPER(MEMNAME) = UPPER("&TabDim");
quit;


/* Desmembra a String armazenada na variável &CmpChaveStg (Campos chave da tabela de Stage)
para identificação dos campos. E concatenação posterior na string do call execute que gerará o código do lookup. */
data WORK.ChaveStg;

CmpChaveStg = &CmpChaveStg;
CmpChaveStgTra = CmpChaveStg;
CmpChaveStgLst = CmpChaveStgTra;
Pos = 1;
PosIni = 1;
PosFim = length(CmpChaveStgTra);
OrdemStg = 0;

do while (Pos ^= 0);
Pos = INDEX(CmpChaveStgTra, ",");
if Pos = 0 then
 CmpChaveStgLst = SUBSTR(CmpChaveStgTra, PosIni, PosFim);
else do;
  CmpChaveStgLst = SUBSTR(CmpChaveStgTra, PosIni, Pos - 1);
  CmpChaveStgTra = SUBSTR(CmpChaveStgTra, Pos+1 , PosFim - Pos);
       PosIni=1;
  PosFim = length(CmpChaveStgTra);
end;
OrdemStg = OrdemStg + 1;
output;
end;

drop CmpChaveStgTra; drop Pos; drop PosIni; drop PosFim;

run;

/* Desmembra a String armazenada na variável &CmpChaveDim (Campos chave da tabela de Dimensão)
para identificação dos campos. E concatenação posterior na string do call execute que gerará o código do lookup. */
data WORK.ChaveDim;

CmpChaveDim = &CmpChaveDim;
CmpChaveDimTra = CmpChaveDim;
CmpChaveDimLst = CmpChaveDim;*AQUI;

Pos = 1;
PosIni = 1;
PosFim = length(CmpChaveDimTra);
OrdemDim = 0;

do while (Pos ^= 0);
Pos = INDEX(CmpChaveDimTra, ",");
if Pos = 0 then
      CmpChaveDimLst = SUBSTR(CmpChaveDimTra, PosIni, PosFim);
else do;
      CmpChaveDimLst = SUBSTR(CmpChaveDimTra, PosIni, Pos - 1);
 CmpChaveDimTra = SUBSTR(CmpChaveDimTra, Pos+1 , PosFim - Pos);
 PosIni=1;
 PosFim = length(CmpChaveDimTra);
end;
OrdemDim = OrdemDim + 1;
output;
end;


drop CmpChaveDimTra; drop Pos; drop PosIni; drop PosFim;

run;


data _null_;

length TamCol $100;
length TxtLengthDim $1000;

declare hash LstTamColDim(dataset:"WORK.TamDim");

LstTamColDim.defineKey("TamCol");
LstTamColDim.defineData("TamCol");
LstTamColDim.defineDone();

declare hiter iterDim('LstTamColDim');

crTamCol = iterDim.first();

do while (crTamCol=0);

TxtLengthDim = CATS(TxtLengthDim, TamCol);
crTamCol = iterDim.next();

end;

drop TamCol; drop crTamCol;

call symput('LengthDim',TxtLengthDim);

run;


/* SCD Tipo I & II */
/* *************** */
data _null_;

/* Tratamento de campos. */
/* ********************* */
/* Definição de padrão para substituição com a função "rxparse". Uso necessário para passagem de
campos de parâmetro separados por vírgula. A substituição de caracteres é feita através da função
"rxchange" e também é feita a limpeza de caracteres brancos através da função "compress". */


rx=rxparse("$',;' to '"",""'"); /* Definição de padrão para substituição. */

* Chave da Tabela de Stage;

length CmpChaveStgCon $1000;
CmpChaveStgCon = &CmpChaveStg || '                                                                                                ';
call rxchange(rx,length(CmpChaveStgCon),CmpChaveStgCon); /* Substituição de valor */
CmpChaveStgCon=compress(CmpChaveStgCon); /* Eliminação de espaços brancos. */

* Campos de retorno da Tabela de Stage;
length CmpRetornoStgCon $1000;
CmpRetornoStgCon = &CmpRetornoStg || '                                                                                                ';
call rxchange(rx,length(CmpRetornoStgCon),CmpRetornoStgCon); /* Substituição de valor */
CmpRetornoStgCon=compress(CmpRetornoStgCon); /* Eliminação de espaços brancos. */

* Chave da Tabela de Dimensão;
length CmpChaveDimCon $1000;
CmpChaveDimCon = &CmpChaveDim || '                                                                                                ';
call rxchange(rx,length(CmpChaveDimCon),CmpChaveDimCon); /* Substituição de valor */
CmpChaveDimCon=compress(CmpChaveDimCon); /* Eliminação de espaços brancos. */


rx=rxparse("$',""' to ' '"); /* Definição de padrão para substituição. */

CS = CmpChaveStgCon;
call rxchange(rx,length(CS),CS); /* Substituição de valor */

CD = CmpChaveDimCon;
call rxchange(rx,length(CD),CD); /* Substituição de valor */


call symput('ChvStg',CS);
call symput('ChvDim',CD);


/* SCD Tipo I & II (Criação de tabela de Update e Insert) */
/* ****************************************************** */
length TxtInUp $5000;

TxtInUp = 'data WORK.SCDI WORK.SCDU; ';

/* O cursor abaixo monta o comando lenght para a tabela de Stage.*/
length TamCol $100;
length TxtLengthStg $1000;

declare hash LstTamColStg(dataset:"WORK.TamStg");

LstTamColStg.defineKey("TamCol");
LstTamColStg.defineData("TamCol");
LstTamColStg.defineDone();

declare hiter iterStg('LstTamColStg');

crTamCol = iterStg.first();

do while (crTamCol=0);

TxtLengthStg = CATS(TxtLengthStg, TamCol);
crTamCol = iterStg.next();

end;

drop TamCol; drop crTamCol;

/* Concatenação dos blocos da string na variável final TxtCodLookup. */
TxtInUp = CATS(TxtInUp, ' ');
TxtInUp = CATS(TxtInUp, TxtLengthStg);


TxtInUp2 =

%if &TipoChv = 'S' %then %do;
symget('CmpChaveSkDim') || ' = ' || symget('Maior') || '; ' ||
'if MISSING(' || symget('CmpChaveSkDim') || ') then ' || symget('CmpChaveSkDim') || ' = 0; ' ||
%end;
'declare hash Stage(dataset:"'|| symget('NomeBiblStg') || '.' || symget('TabStg') ||
'", ordered: "' || TRIM(CmpChaveStgCon) || '"); ' ||
'Stage.defineKey("' || TRIM(CmpChaveStgCon) || '"); ' ||
'Stage.defineData("' || TRIM(CmpRetornoStgCon) || '"); ' ||
'Stage.defineDone(); ' ||
'declare hash Dimensao(dataset:"' || symget('NomeBiblDim') || '.' || symget('TabDim') ||
'", ordered: "' || TRIM(CmpChaveDimCon) || '"); ' ||
'Dimensao.defineKey("' || TRIM(CmpChaveDimCon) || '"); ' ||
'Dimensao.defineData("' || TRIM(CmpChaveDimCon) || '"); ' ||
'Dimensao.defineDone(); ' ||
'declare hiter iterStage("Stage"); ' ||
'rc = iterStage.first(); ' ||
'do while (rc=0); ';

TxtInUp = CATS(TxtInUp, TxtInUp2);


/* O cursor abaixo monta o comando de comparação entre o(s) campo(s) chave da tabela de Dimensão com o(s)
campo(s) chave da tabela de Stage. Para a posterior recuperação de campos feito pelo comando FIND do objeto HASH.
Ex.: CampoChaveTabelaDimensão = CampoChaveTabelaStage; */
length OrdemStg 3; length CmpChaveStgLst $32;

declare hash LstStg(dataset:"WORK.ChaveStg");

LstStg.defineKey("OrdemStg");
LstStg.defineData("OrdemStg","CmpChaveStgLst");
LstStg.defineDone();

declare hiter iterChvStg('LstStg');

length OrdemDim 3; length CmpChaveDimLst $32;

declare hash LstDim(dataset:"WORK.ChaveDim");

LstDim.defineKey("OrdemDim");
LstDim.defineData("OrdemDim","CmpChaveDimLst");
LstDim.defineDone();

cursor = iterChvStg.first();

do while (cursor=0);

OrdemDim = OrdemStg;
cursor = LstDim.find();
TxtInUp = CATS(TxtInUp, CmpChaveDimLst || ' = ' || CmpChaveStgLst || '; ');

cursor = iterChvStg.next();

end;

drop CmpChavePesLst; drop CmpChaveManLst; drop OrdemPes; drop OrdemStg; drop cursor;


TxtInUp3 =
%if &TipoChv = 'S' %then %do;
' if &TipoSCD = 2 then do ' ||

symget('CmpChaveSkDim') || ' = ' || symget('CmpChaveSkDim') || ' + 1; ' ||
' DataIniVig = &DataAtual; ' ||
' DataFimVig = -21914; ' ||
' end; ' ||
%end;

' if Dimensao.FIND() = 0 then ' ||
' output WORK.SCDU; ' ||
' else do; ' ||
%if &TipoChv = 'S' %then %do;
' if &TipoSCD = 1 then ' || symget('CmpChaveSkDim') || ' = ' || symget('CmpChaveSkDim') || ' + 1; ' ||
%end;
' output WORK.SCDI; ' ||
' end; ' ||

' rc = iterStage.next(); ' ||
'end; ' ||

'drop rc; ' ||

'run; ';

TxtInUp = CATS(TxtInUp, TxtInUp3);

/* Gera e executa o código de SCD para criação de tabela de insert e update. */
CALL EXECUTE (TxtInUp);

drop _All_;

run;


%if &TipoSCD = 1 %then %goto Tipo1;
%else %goto Tipo2;

%Tipo1:
/* Preparação Tipo I */
/* ***************** */

data WORK.SCDU;
set WORK.SCDU;
&LengthDim;
drop %if &TipoChv = 'S' %then %do; &CmpChaveSkDim %end; &NomeCmpDatIniVigDim &NomeCmpDatFimVigDim;
run;

data WORK.SCDI;
set WORK.SCDI;
&LengthDim;
drop &NomeCmpDatIniVigDim &NomeCmpDatFimVigDim;
run;

/* Atualização Tipo I */
/* ****************** */

data &NomeBiblDim..&TabDim;
   modify &NomeBiblDim..&TabDim WORK.SCDU;
   by &ChvDim;
   if _iorc_ = %sysrc(_sok) then replace;
   else _ERROR_=0;
run;

%goto Saida;

%Tipo2:
/* Atualização Tipo II */
/* ******************* */

data WORK.SCDData;
set WORK.SCDU;
&LengthDim;
&NomeCmpDatFimVigDim = &DataAtual;
keep &ChvDim &NomeCmpDatFimVigDim;
run;

proc sort data=&NomeBiblDim..&TabDim;
   by &ChvDim descending &NomeCmpDatIniVigDim;
run;

data &NomeBiblDim..&TabDim;
   modify &NomeBiblDim..&TabDim WORK.SCDU;
   by &ChvDim;
   if _iorc_ = %sysrc(_sok) then replace;
   else _ERROR_=0;
run;

/* Inserção Tipo II */
/* **************** */
proc append base=&NomeBiblDim..&TabDim data=WORK.SCDU
force;
run;

%goto Saida;


%Saida:

/* Inserção Tipo I  e II */
/* ********************* */
proc append base=&NomeBiblDim..&TabDim data=WORK.SCDI
force;
run;

/* Ordenação Geral */
/* *************** */
%if &TipoChv = 'S' %then %do;
proc sort data=&NomeBiblDim..&TabDim;
    by &CmpChaveSkDim;
run;
%end;

proc sql;
DROP TABLE WORK.TamStg;
DROP TABLE WORK.TamDim;
DROP TABLE WORK.ChaveStg;
DROP TABLE WORK.ChaveDim;
DROP TABLE Work.SCDI;
DROP TABLE Work.SCDU;
DROP TABLE Work.SCDData;
quit;
%mend;