Usando Expressions dentro do Foreach Loop Container para Captura de Dados de arquivos específicos em um diretório

Na semana passada alguns alunos de um curso de Integration Services que ministrei me trouxeram um questionamento bem interessante sobre um problema com relação à criação de processos de ETL.

Eles trabalham na Caixa Econômica Federal e lá possuem uma situação onde necessitam extrair de um diretório de rede – onde vários arquivos são armazenados de vários processamentos distintos – dados de apenas um arquivo relacionado ao dia útil anterior de um processo específico.

Achei o caso bem interessante e através desse post vou demonstrar como resolver a issue.


Criei uma série de arquivos de para simular o ambiente dos alunos. Nesse diretório teremos dois tipos de arquivos distintos pelo nome que representam dois processos diferentes. são eles [NPD010253.DOC2794.031016007.ARC6576] e [JJX010253.DOC2794.031016007.ARC1076]. Os arquivos são formados pela sigla do processo e uma numeração específica do processo [NPD010253] + o número do documento [DOC2794] + a data dos dados acompanhada de um outro código [031016007] + um outro código e numeração [ARC6576] (toda a nomeclatura do arquivos é meramente demonstrativa).

02

Entendido isso, a problemática é a seguinte: é necessário que desse diretório onde se encontram arquivos de todos os dias úteis do mês de Outubro para ambos os processos (processo [NPD] e processo [JJX]) sejam extraídos somente os dados do arquivo do processo [NPD] porém somente os dados do último dia útil do mês (Lembrando que esse processo hipotético somente será executado de segunda a sexta feira).


Use o Script abaixo para criação do banco de dados de teste que será usado no artigo e acesse o Link para realizar o download do pacote e arquivos usados na demonstração. Para que não ocorram erros, ponham a pasta de arquivos baixados dentro do diretório C:\.

--------------------------------------------
---------- CRIA AMBIENTE DE TESTE ----------
--------------------------------------------
USE master
GO

CREATE DATABASE test_SSIS
GO 

CREATE TABLE tbCliente (
 CodCliente INT,
 NomCLiente VARCHAR(255),
 TipoCliente VARCHAR(55),
 EmailCliente VARCHAR(255),
 TipoTelefone VARCHAR(55),
 TelefoneCliente VARCHAR(55),
 TipoEndereco VARCHAR (55),
 EnderecoCliente VARCHAR (255),
 CodigoPostal VARCHAR (50),
 DatCriacaoRegistro DATETIME
 )

o Pacote nada mais é do que um fluxo de ETL para carga dinâmica de vários arquivos distintos usando o Foreach Loop Container e a parametrização da ConnectionString da conexão com o Flat File Source usando a variável que recebe o nome do arquivos dentro do loop. 

Se ao ler este artigo você ainda não estiver familiarizado com o uso do Foreach Loop Container para carga em loop de vários arquivos em um mesmo destino podem me mandar email que explico como funciona pontualmente já que explicar essa configuração não é a ideia do artigo.

01

A segunda etapa de configuração que será demonstrada abaixo ainda não existe no pacote que está disponível para download. Isto é proposital para que você possa seguir o passo-a-passo de configuração do post.


Para que o ETL carregue apenas os dados do arquivo do último dia útil do mês (sempre o dia útil anterior ao dia atual) e apenas de um dos processos disponíveis no diretório é necessário que, dentro do Foreach Loop Container haja uma parametrização da opção Files: dentro da Aba de configuração Collection.

Para isso, será necessário que você utilize uma Expression. Essa expressão será mapeada no campo Expressions que fica abaixo da seleção de Enumerador do foreach [Enumerator].

03

Dentro da janela de propriedades você irá selecionar em Property a opção FileSpec, na opção Expression, você irá preencher com a expressão abaixo (é possível também, dentro de Property, usar a opção Directory para mapear o diretório do arquivo dinamicamente):

"NPD*" + (
 LEN((DT_WSTR, 2) DAY((DT_DATE) "2016-10-24")) == 1 ? 
 "0" + ((DT_WSTR, 2)(
 DATEPART("DW",(DT_DATE)"2016-10-24") == 2 ? 
 (DT_WSTR, 2)(DAY((DT_DATE)"2016-10-24") -3) :
 (DT_WSTR, 2)(DAY((DT_DATE)"2016-10-24"))
 )
 ) : (
 (DT_WSTR, 2)(DATEPART("DW",(DT_DATE)"2016-10-24") == 2 ? 
 (DT_WSTR, 2)(DAY((DT_DATE)"2016-10-24") -3) :
 (DT_WSTR, 2)(DAY((DT_DATE)"2016-10-24"))
 )
 )
+ (LEN((DT_WSTR, 2) (MONTH((DT_DATE) "2016-10-24"))) == 1 ? 
 "0" + (DT_WSTR, 2) (MONTH((DT_DATE) "2016-10-24")) : 
 (DT_WSTR, 2) (MONTH((DT_DATE) "2016-10-24"))) 
 + (RIGHT((DT_WSTR, 4) (YEAR((DT_DATE) "2016-10-24")),2)) 
 )+ "*.*"

05

Essa expressão irá criar a busca dinâmica pelo nome do arquivo que possua o identificador do processo [NPD] e que tenha data igual ao dia de hoje – 1 (último dia útil).

08

Obs – A expressão acima usa a data ’24-10-2016′ de maneira não dinâmica. Isso está dessa maneira porque o último arquivo do diretório possui a data ’21-10-2016′. Para que em ambiente de produção isso ocorra dinamicamente substitua a data ’24-10-2016′ pela Function GETDATE() conforme a Expression abaixo:

"NPD*" + (
 LEN((DT_WSTR, 2) (DAY(GETDATE()))) == 1 ? 
 "0" + ((DT_WSTR, 2)(
 DATEPART("DW",(GETDATE())) == 2 ? 
 (DT_WSTR, 2)((DAY(GETDATE())) -3) :
 (DT_WSTR, 2)((DAY(GETDATE())))
 )
 ) : (
 (DT_WSTR, 2)(DATEPART("DW",(GETDATE())) == 2 ? 
 (DT_WSTR, 2)(((DAY(GETDATE()))) -3) :
 (DT_WSTR, 2)((DAY(GETDATE())))
 )
 )
+ (LEN((DT_WSTR, 2) (MONTH(GETDATE()))) == 1 ? 
 "0" + (DT_WSTR, 2) (MONTH(GETDATE())) : 
 (DT_WSTR, 2) (MONTH(GETDATE()))) 
 + (RIGHT((DT_WSTR, 4) (YEAR(GETDATE())),2)) 
 )+ "*.*"

Agora, ao executarmos o pacote com o breakpoint habilitado para cada interação do loop poderemos observar que a variável nomeArquivo que, dentro do Flat File Connection, irá se tornar a Connection String para o ETL irá está mapeando o arquivo de nome [NPD010253.DOC2794.211016008.ARC6590.txt] cujo processo corresponde ao processo mapeado na Expression e a data diz respeito à data anterior à data mapeada.

06

Se continuada a execução do pacote, não haverão mais interações de Loop e o pacote irá terminar o processo de carga.

Se você for até o Management Studio e executar a query abaixo, verá que a única data existente dentro da tabela é 2016-10-21.

SELECT DISTINCT CAST(DatCriacaoRegistro 
 AS DATE) AS DatCriacaoRegistro
FROM tbCliente

07


É claro que a Expression deverá ser adaptada para cada ambiente e cada nomeclatura de arquivos segundo as necessidades do processo de carga de ETL.

Espero poder ajudá-los!

Anúncios

Usando Variáveis de Ambientes para Parametrizar conexões no SSIS

Neste post você aprenderá como parametrizar conexões para as bases de dados usando ambientes dentro do SSISDB.

Dessa maneira é possível executar o mesmo pacote em bancos de dados distintos, eliminando assim, a necessidade de alterar o pacote entre ambientes – Desenvolvimento, Homologação e Produção, por exemplo.

IMPORTANTE – Este post somente será válido para quem tem um conhecimento intermediário tanto de SQL Server quanto de Integração de Dados usando o Integration Services.


Em outras ferramentas de Integração de Dados e ETL é possível que você crie um arquivo em um diretório específico com os parâmetros de conexão do Servidor/Banco de Dados, para que, ao executar o pacote, ele capture as credenciais do devido ambiente.

Nas versões anteriores do Integration Services também era possível configurar dessa mesma forma – através de um arquivo .xml – a conexão com os devidos ambientes para execução do pacote.

Porém, para eliminar a possibilidade de erros (um arquivo fica vulnerável à exclusão ou alteração), a partir do SQL Server 2012, é possível criar Ambientes dentro do banco de dados responsável pelo armazenamento dos projetos e pacotes do SSIS, o SSISBD.


Antes de você iniciar o aprendizado, será necessário que você possua a base AdventureWorks2014 devidamente configurada em sua instancia e que você execute o script abaixo. Ele será responsável por criar as bases de dados de AdventureWorks_Homolog e AdventureWorks_Prod.

--Cria a base de dados [AdventureWorks_Homolog]
USE master
GO 

CREATE DATABASE [AdventureWorks_Homolog]
GO

--Cria a a tabela [Person] na base [AdventureWorks_Homolog]
USE [AdventureWorks_Homolog]
GO

CREATE TABLE [dbo].[Person](
 [BusinessEntityID] [int] NULL,
 [PersonType] [nchar](2) NULL,
 [NameStyle] [bit] NULL,
 [Title] [nvarchar](8) NULL,
 [FirstName] [nvarchar](50) NULL,
 [MiddleName] [nvarchar](50) NULL,
 [LastName] [nvarchar](50) NULL,
 [Suffix] [nvarchar](10) NULL,
 [EmailPromotion] [int] NULL,
 [rowguid] [uniqueidentifier] NULL,
 [ModifiedDate] [datetime] NULL
) ON [PRIMARY]
GO

--Cria a base de dados [AdventureWorks_Prod]
Use master
GO

CREATE DATABASE [AdventureWorks_Prod]
GO

--Cria a a tabela [Person] na base [AdventureWorks_Prod]
USE [AdventureWorks_Prod]
GO

CREATE TABLE [dbo].[Person](
 [BusinessEntityID] [int] NULL,
 [PersonType] [nchar](2) NULL,
 [NameStyle] [bit] NULL,
 [Title] [nvarchar](8) NULL,
 [FirstName] [nvarchar](50) NULL,
 [MiddleName] [nvarchar](50) NULL,
 [LastName] [nvarchar](50) NULL,
 [Suffix] [nvarchar](10) NULL,
 [EmailPromotion] [int] NULL,
 [rowguid] [uniqueidentifier] NULL,
 [ModifiedDate] [datetime] NULL
) ON [PRIMARY]
GO

Será necessário também que você crie um projeto, faça o Download do packege que será configurado aqui e adicione esse pacote dentro do projeto que criou.


Agora que o ambiente está pronto, você irá primeiramente criar o SSISDB, dendro da instancia do SQL Server. Para isso, clique com o botão direito em Integration Services Catalog e escolha a opção Create Catalog.

catalog1

Na pagina de configuração deixe tudo default e atribua uma senha para o Catalog.

catalog2

Feito isso, o banco de dados SSISDB estará criado dentro da instancia do SQL Server.

Agora, dentro do projeto do SSIS, você irá criar um project parameter. Para isso, acesse a aba parameter ou, dentro do Solution Explore, acesse Project.params.

01

Crie um parâmetro chamado AdventureWorks_Environment com o tipo de dado String e a opção Required como True. Fazendo isso, você está determinando que esse parâmetro deve ser usado obrigatoriamente quando o pacote do SSIS for executado.

02

Agora que o parâmetro está criado, você irá parametrizar a conexão denominada AdventureWorks_Environment. Para isso, clique com o botão direito na conexão e selecione a opção Parametrize.

03

Na aba de configuração, selecione a opção Use Existing Parameter e escolha o parâmetro que você acabou de criar no passo anterior.

04

O pacote apresentará erro. Isso é normal. O erro ocorre porque essa conexão não tem uma ConnectionString configurada. Isso será realizado daqui a pouco através do SSISDB.

O próximo passo é realizar o deploy desse projeto para dentro do SSISDB que você criou anteriormente. Para isso, na aba de Solution Explore, clique com o botão direito no projeto e depois em Deploy.

07

Selecione o servidor localhost e crie uma pasta para armazenar o projeto dentro do banco de dados. Depois realize o deploy.

Feito isso, o SSISBD passará a armazenar o projeto.

09

O próximo passo será realizar a criação dos dois ambientes dentro do SSISDB. Para isso, selecione com o botão direito a pasta Environment e selecione a opção Create Environment.

10

Crie dois ambientes. Um com o nome Homolog e o outro com o nome Prod conforme a figura abaixo.

11

Abra o primeiro ambiente que foi criado, acesse a aba Variables e crie uma variável de nome AdventureWorks_Environment, com o tipo de dados String e adicione dentro do campo Value a ConnectionString abaixo:


DataSource=.;InitialCatalog=AdventureWorks_Homolog;Provider=SQLNCLI11.1;Integated Security=SSPI;

Faça o mesmo para o ambiente Prod, porém, no campo Value, use a ConnectionString abaixo:


DataSource=.;InitialCatalog=AdventureWorks_Prod;Provider=SQLNCLI11.1;Integated Security=SSPI;

O próximo passo agora é referenciar os dois ambientes no pacote. Você fará isso clicando com o botão direito no Projeto e selecionando a opção Configure. Vá até a aba References e Adicione os dois ambientes que foram criados – Homolog e Prod.

13

Feito isso, navegue até a aba Parameters e, no campo Value do parâmetro do pacote, referencie a Variável que foi criada nos dois ambientes de nome AdventureWorks_Environment.

15

Feito isso, sua aba de configurações ficará igual a imagem abaixo.

14

Depois de criados os ambientes e suas devidas variáveis e de apontar essas variáveis de nome igual para o parâmetro do pacote, você criará dois Jobs para execução da carga. Um de nome Homolog e outro com o nome Prod.

Configure o Step do pacote conforme imagem abaixo:

Job1

Clique na aba Configuration, marque o checkbox Environment e selecione o ambiente Homolog.

job2

Finalize a criação do Job Homolog. Faça o mesmo para a criação do Job Prod, porém, selecione o Ambiente Prod.

Feito isso, execute o Job Homolog. Ao executar um SELECT nas duas tabelas Person – Banco de dados AdventureWorks_Homolog e AdventureWorks_Prod – você perceberá que somente a tabela do banco de homologação recebou dados.

job4

Execute o Job Prod e, ao finalizar a execução, realize os meus dois SELECTs nas tabelas Person das bases de Homologação e Produção. Você perceberá que agora somente a tabela da base AdventureWorks_Prod recebeu dados.

Job5


Essa é a maneira mais simples e segura de trabalhar com o mesmo pacote em vários ambientes, evitando a possibilidade de erros de Deploy ou de execução de teste em um ambiente errado – de Podução, por exemplo.

Construindo um Processo de Carga Dinâmica e Incremental Usando o SSIS

Nesse post você aprenderá como criar um processo na ferramenta de integração de dados da Microsoft Integration Services que efetue a carga de várias tabelas com estruturas e dados diferentes de forma dinâmica


O processo de ETL demonstrado no vídeo abaixo está hoje em funcionamento em um dos nossos clientes da One Way Solution. Ele é responsável pela carga a Stage de Dados de um Data Warehouse. 

Para que seja possível executá-lo, criamos uma stored procedure de nome sp_return_columns que recupera as colunas que serão utilizadas de cada uma das tabelas que participarão do processo. Na Listagem 1 um segue o código caso você queira efetuar testes. 

CREATE PROCEDURE sp_return_columns
	@tablename VARCHAR(2000)
AS
DECLARE @return_column VARCHAR(8000)

;WITH data_capture AS
(
	SELECT STUFF(( SELECT  ',' + SC.name
	FROM sys.tables AS ST
	INNER JOIN sys.columns AS SC
	ON ST.object_id = SC.object_id
	WHERE ST.name = @tablename
	ORDER BY ST.name ASC,SC.name ASC
	FOR XML PATH('')), 1, 1, '') AS ColumnName
)
SELECT @return_column = ColumnName
FROM data_capture

SELECT @return_column AS ColumnName
GO

Listagem 1. Código SQL para criação da stored procedure sp_return_columns.

Para efetuar as cargas dinamicamente, criamos também algumas variáveis e dentro delas utilizamos expressões passando como parâmetro o nome das respectivas tabelas de Origem e Destino e a string de colunas concatenadas que foi disponibilizada através da execução da procedure sp_return_columns

Veja no vídeo como isso fora feito. O processo de carga incremental executa em aproximadamente 3 segundos.

Espero que gostem do vídeo e também poder ajudá-los.