Copiar incrementalmente dados do Azure SQL Database para o Armazenamento de Blobs usando a rastreabilidade de alterações no portal do Azure

APLICÁVEL A: Azure Data Factory Azure Synapse Analytics

Em uma solução de integração de dados, o carregamento incremental de dados após as cargas iniciais de dados é um cenário amplamente utilizado. Os dados alterados dentro de um período na sua loja de dados de origem podem ser facilmente segmentados (por exemplo, LastModifyTime, CreationTime). No entanto, em alguns casos, não há uma maneira explícita de identificar os dados delta desde a última vez que processou os dados. Você pode usar a tecnologia de controle de alterações suportada por armazenamentos de dados, como o Banco de Dados SQL do Azure e o SQL Server, para identificar os dados delta.

Este tutorial descreve como usar o Azure Data Factory com rastreamento de alterações para carregar incrementalmente dados delta do Azure SQL Database no Azure Blob Storage. Para mais informações sobre o acompanhamento de alterações, consulte Acompanhamento de alterações no SQL Server.

Realize os seguintes passos neste tutorial:

Solução de alto nível

Neste tutorial, vais criar duas pipelines que executam as seguintes operações.

1. Carregamento inicial de dados históricos: Crie um pipeline com uma atividade de cópia que copia todos os dados do armazenamento de dados de origem (Azure SQL Database) para o armazenamento de dados de destino (Azure Blob Storage):

   1. Ative a tecnologia de rastreamento de alterações na base de dados de origem no Azure SQL Database.
   2. Obtenha o valor inicial de SYS_CHANGE_VERSION na base de dados para servir como referência ao capturar dados alterados.
   3. Carregue dados completos do banco de dados de origem no Armazenamento de Blobs do Azure.

   

2. Carregamento incremental de dados diferenciais em um cronograma: Você cria uma pipeline com as seguintes atividades e executa-a periodicamente:

   1. Crie duas atividades de pesquisa para obter os valores antigos e novos do SYS_CHANGE_VERSION no Azure SQL Database.

   2. Crie uma atividade de cópia para copiar os dados inseridos, atualizados ou eliminados (os dados delta) entre os dois SYS_CHANGE_VERSION valores do Azure SQL Database para o Azure Blob Storage.

      Carregas os dados delta juntando as chaves primárias das linhas alteradas (entre dois valores SYS_CHANGE_VERSION) de sys.change_tracking_tables com os dados na tabela de origem, e depois moves os dados delta para o destino.

   3. Crie uma atividade de procedimento armazenado para atualizar o valor de SYS_CHANGE_VERSION para a próxima execução do pipeline.

   

Pré-requisitos

• Subscrição do Azure. Se não tiver uma, crie uma conta gratuita antes de começar.
• Banco de Dados SQL do Azure. Utiliza uma base de dados no Azure SQL Database como o armazenamento de dados origem. Se não tiver um, consulte Criar uma base de dados no Azure SQL Database para obter passos para criá-la.
• Conta de armazenamento do Azure. Você usa o Blob Storage como a loja de dados de sink. Se você não tiver uma conta de armazenamento do Azure, consulte Criar uma conta de armazenamento para conhecer as etapas para criar uma. Crie um contêiner chamado adftutorial.

Criar uma tabela de fonte de dados na Azure SQL Database

1. Abra o SQL Server Management Studio e ligue-se à SQL Database.

2. No Server Explorer, clique com o botão direito do rato na sua base de dados e, em seguida, selecione Nova Consulta.

3. Execute o seguinte comando SQL na sua base de dados para criar uma tabela denominada data_source_table como a fonte de dados.

   create table data_source_table
   (
       PersonID int NOT NULL,
       Name varchar(255),
       Age int
       PRIMARY KEY (PersonID)
   );
   INSERT INTO data_source_table
       (PersonID, Name, Age)
   VALUES
       (1, 'aaaa', 21),
       (2, 'bbbb', 24),
       (3, 'cccc', 20),
       (4, 'dddd', 26),
       (5, 'eeee', 22);
   

4. Ative o rastreamento de alterações na sua base de dados e na tabela de origem (data_source_table) executando a seguinte consulta SQL.

   Observação

   • Substitua <your database name> pelo nome do banco de dados no Banco de Dados SQL do Azure que tem data_source_table.
   • Os dados alterados são mantidos por dois dias no exemplo atual. Se carregar os dados alterados a cada três dias ou mais, alguns dados alterados não são incluídos. Necessita de alterar o valor de CHANGE_RETENTION para um número maior ou garantir que o período para carregar os dados alterados seja inferior a dois dias. Para mais informações, consulte Ativar a monitorização de alterações para uma base de dados.

   ALTER DATABASE <your database name>
   SET CHANGE_TRACKING = ON  
   (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON)  
   ALTER TABLE data_source_table
   ENABLE CHANGE_TRACKING  
   WITH (TRACK_COLUMNS_UPDATED = ON)
   

5. Crie uma nova tabela e armazene chamado ChangeTracking_version com um valor padrão executando a seguinte consulta:

   create table table_store_ChangeTracking_version
   (
       TableName varchar(255),
       SYS_CHANGE_VERSION BIGINT,
   );
   DECLARE @ChangeTracking_version BIGINT
   SET @ChangeTracking_version = CHANGE_TRACKING_CURRENT_VERSION();  
   INSERT INTO table_store_ChangeTracking_version
   VALUES ('data_source_table', @ChangeTracking_version)
   

   Observação

   Se os dados não forem alterados após ativar o rastreamento de alterações para o SQL Database, o valor da versão do rastreamento de alterações é 0.

6. Execute a consulta a seguir para criar um procedimento armazenado em seu banco de dados. O pipeline invoca este procedimento armazenado para atualizar a versão de acompanhamento de alterações na tabela que criou no passo anterior.

   CREATE PROCEDURE Update_ChangeTracking_Version @CurrentTrackingVersion BIGINT, @TableName varchar(50)
   AS
   BEGIN
   UPDATE table_store_ChangeTracking_version
   SET [SYS_CHANGE_VERSION] = @CurrentTrackingVersion
   WHERE [TableName] = @TableName
   END    
   

Criar uma fábrica de dados

1. Abra o navegador web Microsoft Edge ou Google Chrome. Atualmente, apenas esses navegadores suportam a interface do usuário (UI) do Data Factory.

2. No portal Azure, no menu à esquerda, selecione Criar um recurso.

3. Selecione Analytics>Data Factory :

   

4. Na página New data factory, insira ADFTutorialDataFactory para o nome.

   O nome da fábrica de dados deve ser globalmente único. Se surgir um erro indicando que o nome escolhido não está disponível, altere o nome (por exemplo, para yournameADFTutorialDataFactory) e tente criar a fábrica de dados novamente. Para obter mais informações, consulte Azure Data Factory naming rules.

5. Selecione a subscrição do Azure onde pretende criar a fábrica de dados.

6. Para Grupo de Recursos, siga um dos seguintes passos:

   • Selecione Usar existente e, em seguida, selecione um grupo de recursos existente na lista suspensa.
   • Selecione Criar novo e, em seguida, introduza o nome de um grupo de recursos.

   Para saber mais sobre grupos de recursos, consulte Utilização de grupos de recursos para gerir os seus recursos do Azure.

7. Para Versão, selecione V2.

8. Em Região, selecione a região para o data factory.

   A lista pendente exibe apenas locais que são suportados. As lojas de dados (por exemplo, Azure Storage e Azure SQL Database) e os recursos de computação (por exemplo, Azure HDInsight) que uma fábrica de dados usa podem estar em outras regiões.

9. Selecione Next: Configuração do Git. Configurar o repositório seguindo as instruções em Método de Configuração 4: Durante a criação da fábrica, ou selecione a opção Configurar Git mais tarde.

10. Selecione Verificar + criar.

11. Selecione Criar.

    No painel, o bloco Deploying Data Factory mostra o status.

    

12. Após a conclusão da criação, a página Data Factory aparecerá. Selecione o mosaico Iniciar estúdio para abrir a interface do Azure Data Factory numa aba separada.

Criar serviços vinculados

Você cria serviços vinculados em uma fábrica de dados para ligar seus repositórios de dados e serviços de computação à fábrica de dados. Nesta secção, cria serviços ligados à sua conta de armazenamento Azure e à sua base de dados na Azure SQL Database.

Criar um serviço de ligação do Azure Storage

Para ligar a sua conta de armazenamento à fábrica de dados:

1. Na interface do usuário do Data Factory, na guia Gerenciar , em Conexões, selecione Serviços vinculados. Em seguida, selecione + Novo ou o botão Criar serviço vinculado .
2. Na janela Novo Serviço Ligado, selecione Azure Blob Storage e, em seguida, clique em Continuar.
3. Insira as seguintes informações:
   1. Em Name, insira AzureStorageLinkedService.
   2. Para Ligar via runtime de integração, selecione o runtime de integração.
   3. Em Tipo de autenticação, selecione um método de autenticação.
   4. Para Nome da conta de armazenamento, selecione sua conta de armazenamento do Azure.
4. Selecione Criar.

Criar um serviço vinculado do Banco de Dados SQL do Azure

Para vincular seu banco de dados ao data factory:

1. Na interface do usuário do Data Factory, na guia Gerenciar , em Conexões, selecione Serviços vinculados. Em seguida, selecione + Novo.

2. Na janela Novo Serviço Vinculado , selecione Banco de Dados SQL do Azure e selecione Continuar.

3. Insira as seguintes informações:

   1. Para Nome, insira AzureSqlDatabaseLinkedService.
   2. Para Nome do Servidor, selecione o seu servidor.
   3. Para Database name, seleccione a sua base de dados.
   4. Em Tipo de autenticação, selecione um método de autenticação. Este tutorial usa autenticação SQL para demonstração.
   5. Para Nome do Utilizador, introduza o nome do utilizador.
   6. Para Palavra-passe, insira uma palavra-passe para o utilizador. Ou forneça as informações para o Azure Key Vault - serviço vinculado AKV, nome secreto e versão secreta.

4. Selecione Testar ligação para testar a ligação.

5. Selecione Criar para criar o serviço vinculado.

   

Criar conjuntos de dados

Nesta seção, você cria conjuntos de dados para representar a fonte de dados e o destino de dados, além do local para armazenar os valores SYS_CHANGE_VERSION.

Criar um conjunto de dados para representar os dados de origem

1. Na interface do utilizador do Data Factory, no separador Autor, selecione o sinal de mais (+). Em seguida, selecione Conjunto de dados ou selecione as reticências para ações do conjunto de dados.

   

2. Selecione Azure SQL Database e depois selecione Continuar.

3. Na janela Definir Propriedades , execute as seguintes etapas:

   1. Em Name, insira SourceDataset.
   2. Para Serviço vinculado, selecione AzureSqlDatabaseLinkedService.
   3. Em Nome da tabela, selecione dbo.data_source_table.
   4. Para Importar esquema, selecione a opção Da ligação/armazenamento.
   5. Selecione OK.

   

Crie um conjunto de dados para representar os dados copiados para o repositório de dados de destino

No procedimento seguinte, você cria um conjunto de dados para representar os dados que são copiados do repositório de dados de origem. Você criou o contêiner adftutorial no Armazenamento de Blobs do Azure como parte dos pré-requisitos. Crie o contêiner se não existir ou defina-o como o nome de um já existente. Neste tutorial, o nome do ficheiro de saída é gerado dinamicamente a partir da expressão @CONCAT('Incremental-', pipeline().RunId, '.txt').

1. Na interface do usuário do Data Factory, na guia Autor , selecione +. Em seguida, selecione Conjunto de dados ou selecione as reticências para ações do conjunto de dados.

   

2. Selecione Azure Blob Storage e, em seguida, selecione Continuar.

3. Selecione o formato do tipo de dados como DelimitedText e, em seguida, selecione Continuar.

4. Na janela Definir propriedades, siga os seguintes passos:

   1. Para Nome, introduza SinkDataset.
   2. Para Serviço ligado, selecione AzureBlobStorageLinkedService.
   3. Em Caminho do arquivo, digite adftutorial/incchgtracking.
   4. Selecione OK.

5. Depois do conjunto de dados aparecer na vista em árvore, vá ao separador Connection e selecione a caixa de texto File name. Quando a opção Adicionar conteúdo dinâmico aparecer, selecione-a.

   

6. A janela do Construtor de Expressões de Pipeline aparece. Cole @concat('Incremental-',pipeline().RunId,'.csv') na caixa de texto.

7. Selecione OK.

Crie um conjunto de dados para representar dados de rastreamento de mudanças.

No seguinte procedimento, cria-se um conjunto de dados para armazenar a versão de acompanhamento de alterações. Criou a tabela table_store_ChangeTracking_version como parte dos pré-requisitos.

1. Na interface do Data Factory, no separador Autor, selecione +, e depois selecione Conjunto de Dados.
2. Selecione Azure SQL Database e depois selecione Continuar.
3. Na janela Definir Propriedades , execute as seguintes etapas:
   1. Para Nome, introduza ChangeTrackingDataset.
   2. Para Serviço vinculado, selecione AzureSqlDatabaseLinkedService.
   3. Para Nome da Tabela, selecione dbo.table_store_ChangeTracking_version.
   4. Para Importar esquema, selecione a opção Da ligação/armazenamento.
   5. Selecione OK.

Criar um fluxo de trabalho para a cópia completa

No procedimento seguinte, cria-se um pipeline com uma atividade de cópia que replica todos os dados do repositório de dados de origem (Base de Dados SQL do Azure) para o repositório de dados de destino (Armazenamento de Blobs do Azure).

1. Na interface do utilizador do Data Factory, no separador Author, selecione + e, em seguida, selecione Pipeline>Pipeline.

   

2. Uma nova aba aparece para configurar o pipeline. A pipeline também aparece na vista em árvore. Na janela Propriedades, altere o nome da pipeline para FullCopyPipeline.

3. Na caixa de ferramentas Atividades , expanda Mover & transformar. Tome um dos seguintes passos:

   • Arraste a atividade de cópia para a superfície do designer de pipeline.
   • Na barra de pesquisa, em Atividades, procure a atividade de cópia de dados e defina o nome como FullCopyActivity.

4. Altere para a aba Source. Para Source Dataset, selecione SourceDataset.

5. Alterne para a guia Coletor . Em Sink Dataset, selecione SinkDataset.

6. Para validar a definição da pipeline, selecione Validate na barra de ferramentas. Confirme que não há erro de validação. Feche a saída de validação do pipeline.

7. Para publicar entidades (serviços vinculados, conjuntos de dados e pipelines), selecione Publicar tudo. Aguarde até surgir a mensagem Publicação bem-sucedida.

   

8. Para ver as notificações, selecione o botão Mostrar notificações .

Executar o pipeline de cópia completa

1. Na interface do Data Factory, na barra de ferramentas para o pipeline, selecione Adicionar gatilho e, em seguida, selecione Acionar agora.

   

2. Na janela Execução do pipeline, selecione OK.

   

Monitorar o pipeline de cópia completa

1. Na interface do Data Factory, selecione o separador Monitor. A execução do pipeline e o seu estado aparecem na lista. Para atualizar a lista, selecione Atualizar. Passe o cursor sobre a execução do pipeline para obter a opção Reexecutar ou Consumo.

   

2. Para ver as execuções de atividade associadas à execução do pipeline, selecione o nome do pipeline na coluna Nome do Pipeline. Há apenas uma atividade no pipeline, então há apenas uma entrada na lista. Para regressar à visualização das execuções do pipeline, selecione o link Todas as execuções do pipeline no topo.

Revise os resultados

A pasta incchgtracking do contêiner adftutorial inclui um arquivo chamado incremental-<GUID>.csv.

O ficheiro deve conter os dados da sua base de dados:

PersonID,Name,Age
1,"aaaa",21
2,"bbbb",24
3,"cccc",20
4,"dddd",26
5,"eeee",22

5,eeee,PersonID,Name,Age
1,"aaaa",21
2,"bbbb",24
3,"cccc",20
4,"dddd",26
5,"eeee",22

Adicione mais dados à tabela de origem

Execute a seguinte consulta na sua base de dados para adicionar uma linha e atualizar uma linha.

INSERT INTO data_source_table
(PersonID, Name, Age)
VALUES
(6, 'new','50');


UPDATE data_source_table
SET [Age] = '10', [name]='update' where [PersonID] = 1

Criar um pipeline para a cópia delta

No procedimento seguinte, cria-se uma cadeia de atividades e executa-se periodicamente. Quando executa o pipeline:

• As atividades de pesquisa obtêm os valores antigos e novos da base de dados SQL do Azure e passam-nos para a atividade de cópia.
• A atividade de cópia copia os dados inseridos, atualizados ou eliminados entre os dois valores SYS_CHANGE_VERSION do Azure SQL Database para o Azure Blob Storage.
• A atividade de procedimento armazenado atualiza o valor de SYS_CHANGE_VERSION para a próxima execução do pipeline.

1. No interface element needs translation, but we translate auxiliary text for a consistent experience with the Portuguese locale. The improved sentence should be: Na interface do Data Factory, mude para o separador Autor. Selecione +, e depois selecione Pipeline>Pipeline.

   

2. Uma nova aba aparece para configurar o pipeline. A pipeline também aparece na vista em árvore. Na janela de Propriedades, altere o nome do pipeline para PipelineDeCópiaIncremental.

3. Expanda Geral na caixa de ferramentas Atividades. Arraste a atividade de pesquisa para a superfície do designer de pipeline ou procure na caixa Pesquisar atividades. Defina o nome da atividade para LookupLastChangeTrackingVersionActivity. Esta atividade obtém a versão de controle de alterações usada na última operação de cópia armazenada na tabela table_store_ChangeTracking_version.

4. Mude para o separador Definições na janela Propriedades. Para Source Dataset, selecione ChangeTrackingDataset.

5. Arraste a atividade de pesquisa da caixa de ferramentas de Atividades para a superfície do designer de pipeline. Defina o nome da atividade para LookupCurrentChangeTrackingVersionActivity. Esta atividade obtém a versão atual de rastreamento de mudanças.

6. Altere para o separador Definições na janela Propriedades e, em seguida, siga os seguintes passos:

   1. Para Conjunto de dados de origem, selecione ConjuntoDeDadosDeOrigem.

   2. Para Usar consulta, selecione Consulta.

   3. Para Consulta, insira a seguinte consulta SQL:

      SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion
      

   

7. Na caixa de ferramentas Atividades , expanda Mover & transformar. Arraste a atividade de cópia de dados para a superfície do designer de pipeline. Defina o nome da atividade para IncrementalCopyActivity. Esta atividade copia os dados entre a última versão de rastreamento de alterações e a versão atual de rastreamento de alterações para o repositório de dados de destino.

8. Mude para o separador Fonte na janela de Propriedades e, em seguida, siga os passos seguintes:

   1. Para Conjunto de dados de origem, selecione ConjuntoDeDadosDeOrigem.

   2. Para Usar consulta, selecione Consulta.

   3. Para Consulta, insira a seguinte consulta SQL:

      SELECT data_source_table.PersonID,data_source_table.Name,data_source_table.Age, CT.SYS_CHANGE_VERSION, SYS_CHANGE_OPERATION from data_source_table RIGHT OUTER JOIN CHANGETABLE(CHANGES data_source_table, @{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.SYS_CHANGE_VERSION}) AS CT ON data_source_table.PersonID = CT.PersonID where CT.SYS_CHANGE_VERSION <= @{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}
      

   

9. Alterne para a guia Coletor . Em Sink Dataset, selecione SinkDataset.

10. Conecte ambas as atividades de pesquisa à atividade de cópia uma de cada vez. Arraste o botão verde que está ligado à atividade de busca para a atividade de cópia.

11. Arraste a atividade de procedimento armazenado da caixa de ferramentas Atividades para a superfície do designer de pipeline. Defina o nome da atividade para StoredProceduretoUpdateChangeTrackingActivity. Esta atividade atualiza a versão de controlo de alterações na table_store_ChangeTracking_version tabela.

12. Mude para o separador Definições e, em seguida, siga os seguintes passos:

    1. Para Serviço vinculado, selecione AzureSqlDatabaseLinkedService.
    2. Para Nome do procedimento armazenado, selecione Update_ChangeTracking_Version.
    3. Selecione Importar.
    4. Na secção Parâmetros do procedimento armazenado, especifique os seguintes valores para os parâmetros:

    Nome	Tipo	Valor
    CurrentTrackingVersion	Int64	@{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}
    TableName	Cordão	@{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.TableName}

    

13. Conecte a atividade de cópia à atividade de procedimento armazenado. Arraste o botão verde que está ligado à atividade de cópia para a atividade de procedimento armazenado.

14. Selecione Validate na barra de ferramentas. Confirme que não existem erros de validação. Feche a janela Relatório de Validação de Pipeline .

15. Publique entidades (serviços associados, conjuntos de dados e pipelines) para o serviço Data Factory selecionando o botão Publicar tudo. Espere até que a mensagem Publishing succeeded apareça.

    

Executar o pipeline de cópia incremental

1. Selecione Adicionar gatilho na barra de ferramentas do pipeline e, em seguida, selecione Acionar agora.

   

2. Na janela Pipeline Run, selecione OK.

Monitorizar a canalização incremental de cópias

1. Selecione o separador Monitor. A execução do pipeline e o seu estado aparecem na lista. Para atualizar a lista, selecione Atualizar.

   

2. Para exibir as execuções de atividade associadas à execução do pipeline, selecione o link IncrementalCopyPipeline na coluna Nome do pipeline . As execuções de atividade aparecem numa lista.

   

Revise os resultados

O segundo arquivo aparece na pasta incchgtracking do contêiner adftutorial .

O ficheiro deve conter apenas os dados delta da sua base de dados. O registro com U é a linha atualizada no banco de dados e I é a linha adicionada.

PersonID,Name,Age,SYS_CHANGE_VERSION,SYS_CHANGE_OPERATION
1,update,10,2,U
6,new,50,1,I

As três primeiras colunas são dados alterados de data_source_table. As últimas duas colunas são os metadados da tabela para o sistema de rastreamento de alterações. A quarta coluna é o valor SYS_CHANGE_VERSION para cada linha alterada. A quinta coluna é a operação: U = atualizar, I = inserir. Para obter detalhes sobre as informações de rastreamento de alterações, consulte CHANGETABLE.

==================================================================
PersonID Name    Age    SYS_CHANGE_VERSION    SYS_CHANGE_OPERATION
==================================================================
1        update  10            2                                 U
6        new     50	       1                            	 I

Conteúdo relacionado

Avance para o tutorial a seguir para aprender a copiar apenas arquivos novos e alterados, com base em LastModifiedDate: