Migrar dados do Amazon S3 para Azure Data Lake Storage Gen2

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

Use os templates para migrar petabytes de dados que consistem em centenas de milhões de ficheiros do Amazon S3 para o Azure Data Lake Storage Gen2.

Sobre os modelos de solução

A partição de dados é recomendada especialmente ao migrar mais de 10 TB de dados. Para particionar os dados, aproveite a configuração 'prefixo' para filtrar as pastas e arquivos no Amazon S3 por nome e, em seguida, cada trabalho de cópia do ADF pode copiar uma partição de cada vez. Você pode executar vários trabalhos de cópia do ADF simultaneamente para um melhor desempenho.

A migração de dados normalmente requer uma migração histórica única mais a sincronização periódica das alterações do AWS S3 para o Azure. Existem dois modelos abaixo, onde um modelo cobre a migração histórica única de dados e outro aborda a sincronização das alterações do AWS S3 para o Azure.

Para o modelo de migração de dados históricos do Amazon S3 para o Azure Data Lake Storage Gen2

Este modelo (nome template: migrar dados históricos do AWS S3 para Azure Data Lake Storage Gen2) assume que escreveu uma lista de partições numa tabela de controlo externa em Base de Dados SQL do Azure. Assim, ele usará uma atividade Lookup para recuperar a lista de partições da tabela de controle externo, iterar sobre cada partição e assegurar que cada trabalho de cópia do ADF copie uma partição de cada vez. Uma vez concluído qualquer trabalho de cópia, ele utiliza o Procedimento Armazenado para atualizar o estado de cópia de cada partição na tabela de controlo.

O modelo contém cinco atividades:

• Lookup recupera as partições que não foram copiadas para Azure Data Lake Storage Gen2 a partir de uma tabela de controlo externa. O nome da tabela é s3_partition_control_table e a consulta para carregar dados da tabela é "SELECT PartitionPrefix FROM s3_partition_control_table WHERE SuccessOrFailure = 0".
• ForEach obtém a lista de partições da atividade Pesquisa e itera cada partição para a atividade TriggerCopy . Você pode definir o batchCount para executar vários trabalhos de cópia do ADF simultaneamente. Nós definimos 2 neste modelo.
• ExecutePipeline executa CopyFolderPartitionFromS3 pipeline. O motivo pelo qual criamos outro pipeline para fazer com que cada trabalho de cópia copie uma partição é porque isso facilitará a reexecução do trabalho de cópia com falha para recarregar essa partição específica novamente a partir do AWS S3. Todos os outros trabalhos de cópia que carregam outras partições não serão afetados.
• Copy copia cada partição do AWS S3 para Azure Data Lake Storage Gen2.
• SqlServerStoredProcedure atualiza o status da cópia de cada partição na tabela de controle.

O modelo contém dois parâmetros:

• AWS_S3_bucketName é o nome do bucket no AWS S3 de onde você deseja migrar dados. Se quiser migrar dados de vários buckets no AWS S3, você pode adicionar mais uma coluna na tabela de controle externo para armazenar o nome do bucket para cada partição e também atualizar o pipeline para recuperar dados dessa coluna de acordo.
• Azure_Storage_fileSystem é o nome do teu sistema de ficheiros em Azure Data Lake Storage Gen2 para onde queres migrar os dados.

Para o modelo copiar apenas ficheiros alterados do Amazon S3 para o Azure Data Lake Storage Gen2

Este modelo (nome template: copy delta data from AWS S3 to Azure Data Lake Storage Gen2) utiliza o LastModifiedTime de cada ficheiro para copiar apenas os ficheiros novos ou atualizados do AWS S3 para Azure. Esteja ciente de que se seus arquivos ou pastas já foram particionados por tempo com informações de timelice como parte do nome do arquivo ou pasta no AWS S3 (por exemplo, /aaaa/mm/dd/file.csv), você pode ir para este tutorial para obter a abordagem mais eficiente para o carregamento incremental de novos arquivos. Este modelo assume que escreveu uma lista de partições numa tabela de controlo externa no Base de Dados SQL do Azure. Assim, ele usará uma atividade Lookup para recuperar a lista de partições da tabela de controle externo, iterar sobre cada partição e assegurar que cada trabalho de cópia do ADF copie uma partição de cada vez. Quando cada trabalho de cópia começa a copiar os arquivos do AWS S3, ele depende da propriedade LastModifiedTime para identificar e copiar somente os arquivos novos ou atualizados. Uma vez concluído qualquer trabalho de cópia, ele utiliza o Procedimento Armazenado para atualizar o estado de cópia de cada partição na tabela de controlo.

O modelo contém sete atividades:

• A pesquisa recupera as partições de uma tabela de controle externo. O nome da tabela é s3_partition_delta_control_table e a consulta para carregar dados da tabela é "select distinct PartitionPrefix from s3_partition_delta_control_table".
• ForEach obtém a lista de partições da atividade Pesquisa e itera cada partição para a atividade TriggerDeltaCopy . Você pode definir o batchCount para executar vários trabalhos de cópia do ADF simultaneamente. Nós definimos 2 neste modelo.
• ExecutePipeline executa o pipeline DeltaCopyFolderPartitionFromS3. O motivo pelo qual criamos outro pipeline para fazer com que cada trabalho de cópia copie uma partição é porque isso facilitará a reexecução do trabalho de cópia com falha para recarregar essa partição específica novamente a partir do AWS S3. Todos os outros trabalhos de cópia que carregam outras partições não serão afetados.
• Lookup recupera a hora da última execução do trabalho de cópia a partir da tabela de controle externo, para que os arquivos novos ou atualizados possam ser identificados através de LastModifiedTime. O nome da tabela é s3_partition_delta_control_table e a consulta para carregar dados da tabela é "select max(JobRunTime) as LastModifiedTime from s3_partition_delta_control_table onde PartitionPrefix = '@{pipeline().parameters.prefixStr}' e SuccessOrFailure = 1".
• Copy copia apenas ficheiros novos ou alterados para cada partição do AWS S3 para Azure Data Lake Storage Gen2. A propriedade modifiedDatetimeStart é especificada como o tempo de execução do último trabalho de cópia. A propriedade modifiedDatetimeEnd é definida para a hora da execução do trabalho de cópia atual. Esteja ciente de que a hora é aplicada ao fuso horário UTC.
• SqlServerStoredProcedure atualiza o status da cópia de cada partição e copia o tempo de execução na tabela de controle quando ela for bem-sucedida. A coluna de SuccessOrFailure é definida como 1.
• SqlServerStoredProcedure atualiza o status de cópia de cada partição e copia o tempo de execução na tabela de controle quando ela falha. A coluna de SuccessOrFailure é definida como 0.

O modelo contém dois parâmetros:

• AWS_S3_bucketName é o nome do bucket no AWS S3 de onde você deseja migrar dados. Se quiser migrar dados de vários buckets no AWS S3, você pode adicionar mais uma coluna na tabela de controle externo para armazenar o nome do bucket para cada partição e também atualizar o pipeline para recuperar dados dessa coluna de acordo.
• Azure_Storage_fileSystem é o nome do teu sistema de ficheiros em Azure Data Lake Storage Gen2 para onde queres migrar os dados.

Como usar esses dois modelos de solução

Para o modelo de migração de dados históricos do Amazon S3 para o Azure Data Lake Storage Gen2

1. Crie uma tabela de controlo no Base de Dados SQL do Azure para armazenar a lista de partições do AWS S3.

   Nota

   O nome da tabela é s3_partition_control_table. O esquema da tabela de controlo é PartitionPrefix e SuccessOrFailure, onde PartitionPrefix é a definição de prefixo em S3 para filtrar as pastas e ficheiros na Amazon S3 por nome, e SuccessOrFailure é o estado de cópia de cada partição: 0 significa que esta partição não foi copiada para o Azure e 1 significa que esta partição foi copiada com sucesso para o Azure. Existem 5 partições definidas na tabela de controle e o status padrão de copiar cada partição é 0.

   CREATE TABLE [dbo].[s3_partition_control_table](
       [PartitionPrefix] [varchar](255) NULL,
       [SuccessOrFailure] [bit] NULL
   )
   
   INSERT INTO s3_partition_control_table (PartitionPrefix, SuccessOrFailure)
   VALUES
   ('a', 0),
   ('b', 0),
   ('c', 0),
   ('d', 0),
   ('e', 0);
   

2. Crie um Procedimento Armazenado na mesma Base de Dados SQL do Azure para a tabela de controlo.

   Nota

   O nome do procedimento armazenado é sp_update_partition_success. Ele será invocado pela atividade SqlServerStoredProcedure em seu pipeline do ADF.

   CREATE PROCEDURE [dbo].[sp_update_partition_success] @PartPrefix varchar(255)
   AS
   BEGIN
   
       UPDATE s3_partition_control_table
       SET [SuccessOrFailure] = 1 WHERE [PartitionPrefix] = @PartPrefix
   END
   GO
   

3. Vai ao template Migrar dados históricos do AWS S3 para Azure Data Lake Storage Gen2. Introduza as ligações na sua tabela de controlo externa, AWS S3 como armazenamento de dados de origem e Azure Data Lake Storage Gen2 como armazenamento de dados de destino. Esteja ciente de que a tabela de controlo externo e o procedimento armazenado referem-se à mesma conexão.

   

4. Selecione Utilizar este modelo.

   

5. Você vê que os 2 pipelines e 3 conjuntos de dados foram criados, conforme mostrado no exemplo a seguir:

   

6. Vá para o pipeline "BulkCopyFromS3" e selecione Depurar, introduza os parâmetros. Em seguida, selecione Concluir.

   

7. Você vê resultados semelhantes ao exemplo a seguir:

   

Para o modelo copiar apenas ficheiros alterados do Amazon S3 para o Azure Data Lake Storage Gen2

1. Crie uma tabela de controlo no Base de Dados SQL do Azure para armazenar a lista de partições do AWS S3.

   Nota

   O nome da tabela é s3_partition_delta_control_table. O esquema da tabela de controlo é PartitionPrefix, JobRunTime e SuccessOrFailure, onde PartitionPrefix é a definição de prefixo em S3 para filtrar pastas e ficheiros na Amazon S3 por nome, JobRunTime é o valor datatime quando trabalhos de cópia são executados, e SucessoOrFailure é o estado de cópia de cada partição: 0 significa que esta partição não foi copiada para o Azure e 1 significa que esta partição foi copiada para o Azure com sucesso. Existem 5 partições definidas na tabela de controle. O valor padrão para JobRunTime pode ser o momento em que a migração de dados históricos única é iniciada. A atividade de cópia do ADF copiará os arquivos no AWS S3 que foram modificados pela última vez após esse período. O status padrão de copiar cada partição é 1.

   CREATE TABLE [dbo].[s3_partition_delta_control_table](
       [PartitionPrefix] [varchar](255) NULL,
       [JobRunTime] [datetime] NULL,
       [SuccessOrFailure] [bit] NULL
       )
   
   INSERT INTO s3_partition_delta_control_table (PartitionPrefix, JobRunTime, SuccessOrFailure)
   VALUES
   ('a','1/1/2019 12:00:00 AM',1),
   ('b','1/1/2019 12:00:00 AM',1),
   ('c','1/1/2019 12:00:00 AM',1),
   ('d','1/1/2019 12:00:00 AM',1),
   ('e','1/1/2019 12:00:00 AM',1);
   

2. Crie um Procedimento Armazenado na mesma Base de Dados SQL do Azure para a tabela de controlo.

   Nota

   O nome do procedimento armazenado é sp_insert_partition_JobRunTime_success. Ele será invocado pela atividade SqlServerStoredProcedure em seu pipeline do ADF.

   CREATE PROCEDURE [dbo].[sp_insert_partition_JobRunTime_success] @PartPrefix varchar(255), @JobRunTime datetime, @SuccessOrFailure bit
   AS
   BEGIN
       INSERT INTO s3_partition_delta_control_table (PartitionPrefix, JobRunTime, SuccessOrFailure)
       VALUES
           (@PartPrefix,@JobRunTime,@SuccessOrFailure)
   END
   GO
   

3. Vai ao modelo Copiar os dados delta do AWS S3 para Azure Data Lake Storage Gen2. Introduza as ligações na sua tabela de controlo externa, com AWS S3 como a loja de origem de dados e Azure Data Lake Storage Gen2 como a loja de destino. Esteja ciente de que a tabela de controlo externo e o procedimento armazenado referem-se à mesma conexão.

   

4. Selecione Utilizar este modelo.

   

5. Você vê que os 2 pipelines e 3 conjuntos de dados foram criados, conforme mostrado no exemplo a seguir:

   

6. Vá para o pipeline "DeltaCopyFromS3", selecione Depurar e insira os Parâmetros. Em seguida, selecione Concluir.

   

7. Você vê resultados semelhantes ao exemplo a seguir:

   

8. Você também pode verificar os resultados da tabela de controle por uma consulta "selecionar * de s3_partition_delta_control_table", você verá a saída semelhante ao exemplo a seguir:

   

Conteúdos relacionados

• Copiar arquivos de vários contêineres
• Mover ficheiros