"Exercício: Criar um fluxo de dados de mapeamento do Azure Data Factory"
Como transformar dados com o Fluxo de Dados de Mapeamento
Você pode executar nativamente transformações de dados sem código do Azure Data Factory usando a tarefa Fluxo de Dados de Mapeamento. Os Fluxos de Dados de Mapeamento fornecem uma experiência totalmente visual sem a necessidade de codificação. Os seus fluxos de dados serão executados no seu cluster em execução para o processamento de dados expandidos. As atividades de fluxo de dados podem ser operacionalizadas por meio das funcionalidades existentes de agendamento, de controle, de fluxo e de monitoramento do Data Factory.
Ao criar fluxos de dados, você pode habilitar o modo de depuração, que ativa um pequeno cluster interativo do Spark. Ative o modo de depuração alternando o controle deslizante na parte superior do módulo de criação. Os clusters de depuração levam alguns minutos para serem ativados, mas podem ser usados para visualizar de modo interativo a saída da sua lógica de transformação.
Com o Fluxo de Dados de Mapeamento adicionado e o cluster do Spark em execução, será possível realizar a transformação e executar e visualizar os dados. Nenhuma codificação é exigida, pois o Azure Data Factory administra toda a conversão de código, a otimização de caminho e a execução dos seus trabalhos de fluxo de dados.
Como adicionar dados de origem ao Fluxo de Dados de Mapeamento
Abra a tela Fluxo de Dados de Mapeamento. Clique no botão Adicionar Fonte na tela Fluxo de Dados. No menu suspenso do conjunto de dados de origem, selecione a fonte de dados. Neste exemplo, o conjunto de dados do ADLS Gen2 é usado
Há alguns pontos a serem observados:
- Se o seu conjunto de dados estiver apontando para uma pasta com outros arquivos e você quiser usar apenas um arquivo, talvez seja necessário criar outro conjunto de dados ou utilizar a parametrização para garantir que apenas um arquivo específico seja lido
- Se você não tiver importado o seu esquema no ADLS, mas já tiver ingerido os seus dados, acesse a guia 'Esquema' do conjunto de dados e clique em 'Importar esquema' para que o seu fluxo de dados conheça a projeção do esquema.
O Fluxo de Dados de Mapeamento segue uma abordagem de ELT (extração, carregamento, transformação) e funciona com conjuntos de dados de preparo que estejam no Azure. Atualmente, os seguintes conjuntos de dados podem ser usados em uma transformação de origem:
- Armazenamento de Blobs do Azure (JSON, Avro, Text, Parquet)
- Azure Data Lake Storage Gen1 (JSON, Avro, Text, Parquet)
- Azure Data Lake Storage Gen2 (JSON, Avro, Text, Parquet)
- Azure Synapse Analytics
- Banco de Dados SQL do Azure
- Azure Cosmos DB
O Azure Data Factory tem acesso a mais de 80 conectores nativos. Para incluir dados de outras fontes no seu fluxo de dados, use a atividade Copy para carregar esses dados em uma das áreas de preparo com suporte.
Quando o cluster de depuração estiver aquecido, verifique se os dados estão carregados corretamente na guia Visualização de Dados. Depois de clicar no botão Atualizar, o Fluxo de Dados de Mapeamento exibirá um instantâneo da aparência dos seus dados em cada transformação.
Como usar transformações no Fluxo de Dados de Mapeamento
Agora que você moveu os dados para o Azure Data Lake Storage Gen2, está pronto para criar um Fluxo de Dados de Mapeamento que transformará os seus dados em escala por meio de um cluster do Spark e então os carregará em um Data Warehouse.
As principais tarefas para isso são as seguintes:
Preparar o ambiente
Adicionando uma fonte de dados
Usar a transformação de Fluxo de Dados de Mapeamento
Gravar em um coletor de dados
Tarefa 1: Preparar o ambiente
Ativar a Depuração do Fluxo de Dados: ative o controle deslizante Depuração do Fluxo de Dados localizado na parte superior do módulo de criação.
Observação
Os clusters do Fluxo de Dados levam de cinco a sete minutos para serem ativados.
Adicione uma atividade Data Flow. No painel Atividades, abra o acordeão Mover e Transformar e arraste a atividade Data Flow para a tela do pipeline. Na folha que aparece, clique em Criar Fluxo de Dados e selecione Fluxo de Dados de Mapeamento e OK. Clique na guia pipeline1 e arraste a caixa verde da atividade Copy para a atividade Data Flow para criar uma condição de sucesso. Você verá o seguinte na tela:
Tarefa 2: Adicionar uma fonte de dados
Adicione uma origem do ADLS. Clique duas vezes no objeto Fluxo de Dados de Mapeamento na tela. Clique no botão Adicionar Fonte na tela Fluxo de Dados. No menu suspenso Conjunto de dados de origem, selecione o conjunto de dados ADLSG2 usado na sua atividade Copy
- Se o seu conjunto de dados estiver apontando para uma pasta com outros arquivos, talvez seja necessário criar outro conjunto de dados ou utilizar a parametrização para garantir que apenas o arquivo moviesDB.csv seja lido
- Se você não tiver importado o seu esquema no ADLS, mas já tiver ingerido os seus dados, acesse a guia 'Esquema' do conjunto de dados e clique em 'Importar esquema' para que o seu fluxo de dados conheça a projeção do esquema.
Quando o cluster de depuração estiver aquecido, verifique se os dados estão carregados corretamente na guia Visualização de Dados. Depois de clicar no botão Atualizar, o Fluxo de Dados de Mapeamento exibirá um instantâneo da aparência dos seus dados em cada transformação.
Tarefa 3: Usar a transformação de Fluxo de Dados de Mapeamento
Adicione uma transformação Selecionar para renomear e remover uma coluna. Na visualização dos dados, talvez você tenha notado que a coluna "Rotton Tomatoes" tem um erro de ortografia. Para nomeá-la corretamente e remover a coluna Classificação não usada, você pode adicionar uma transformação Selecionar clicando no ícone + ao lado do nó de origem do ADLS e escolhendo Selecionar no Modificador de esquema.
No campo Nome, altere 'Rotton' para 'Rotten'. Para remover a coluna Classificação, passe o mouse sobre ela e clique no ícone de lixeira.
Adicione uma transformação de filtro para filtrar anos indesejados. Digamos que você esteja interessado apenas em filmes lançados após 1951. Você pode adicionar uma Transformação de filtro para especificar uma condição de filtro clicando no ícone + ao lado da transformação Selecionar e escolhendo Filtrar em Modificador de Linha. Clique na caixa de expressão para abrir o Construtor de Expressões e insira a condição de filtro. Usar a sintaxe da linguagem de expressão do Fluxo de Dados de Mapeamento, toInteger(year) > 1950 converte o valor de ano da cadeia de caracteres em um inteiro e filtra as linhas se esse valor está acima de 1950.
Você pode usar o painel de visualização de dados inserido do Construtor de Expressões para verificar se a condição está funcionando corretamente
Adicione uma transformação Derivar para calcular o gênero principal. Como você deve ter notado, a coluna de gêneros é uma cadeia de caracteres delimitada por um caractere '|'. Se você se preocupa apenas com o primeiro gênero em cada coluna, pode derivar uma nova coluna chamada PrimaryGenre por meio da transformação de Coluna Derivada clicando no ícone + ao lado da sua transformação de Filtro e escolhendo Derivado no Modificador de Esquema. Semelhante à transformação de filtro, a coluna derivada usa o Construtor de Expressões do Fluxo de Dados de Mapeamento para especificar os valores da nova coluna.
Nesse cenário, você está tentando extrair o primeiro gênero da coluna de gêneros, que é formatada como 'genre1|genre2|...|genreN'. Use a função locate para obter o primeiro índice baseado em um do '|' na cadeia de caracteres de gêneros. Usando a função iif, se esse índice for maior que 1, o gênero principal poderá ser calculado por meio da função left, que retorna todos os caracteres em uma cadeia de caracteres para a esquerda de um índice. Caso contrário, o valor PrimaryGenre será igual ao campo de gêneros. Você pode verificar a saída por meio do painel Visualização de dados do Construtor de Expressões.
Classifique os filmes por meio de uma Transformação de Janela. Digamos que você esteja interessado na classificação de um filme no ano de lançamento e no gênero específico. Você pode adicionar uma transformação de Janela para definir as agregações baseadas em janela clicando no ícone + ao lado da transformação de Coluna Derivada e clicando em Janela no Modificador de esquema. Para fazer isso, especifique sobre o que você está criando janelas, o que você está classificando, qual é o intervalo e como calcular as suas novas colunas de janela. Neste exemplo, vamos criar uma janela em PrimaryGenre e year com um intervalo não associado, classificar no Rotten Tomato por ordem decrescente e calcular uma nova coluna chamada RatingsRank que será igual à classificação que cada filme tem no gênero/ano específico.
Agregar classificações com uma transformação Agregar. Agora que você reuniu e derivou todos os seus dados necessários, podemos adicionar uma transformação de Agregação para calcular métricas com base em um grupo desejado clicando no ícone + ao lado da transformação de Janela e clicando em Agregar no Modificador de Esquema. Como você fez na transformação de janela, vamos agrupar filmes por PrimaryGenre e year
Na guia Agregações, é possível criar agregações calculadas sobre o grupo especificado por colunas. Para cada gênero e ano, vamos obter a classificação média do Rotten Tomatoes, o filme com a maior e menor classificação (utilizando a função de criação de janela) e o número de filmes que há em cada grupo. A agregação reduz significativamente o número de linhas no seu fluxo de transformação e propaga apenas as colunas group by e aggregate especificadas na transformação.
- Para ver como a transformação de agregação altera os seus dados, use a guia Visualização de Dados
Especifique a condição Upsert por meio de uma transformação de Alteração de Linha. Se estiver gravando em um coletor tabular, você poderá especificar as políticas de inserção, exclusão, atualização e de upsert em linhas usando a transformação de Alteração de Linha clicando no ícone + ao lado de sua transformação de Agregação e clicando em Alterar Linha no Modificador de linha. Como você está sempre inserindo e atualizando, será possível especificar que será executado upsert em todas as linhas.
Tarefa 4: Como gravar em um coletor de dados
- Grave em um coletor do Azure Synapse Analytics. Agora que você concluiu toda a sua lógica de transformação, você está pronto para gravar em um Coletor.
Adicione um Coletor clicando no ícone + ao lado de sua transformação de Upsert e clicando no Coletor em Destino.
Na guia Coletor, crie um conjunto de dados do data warehouse usando o botão + Novo.
Selecione Azure Synapse Analytics na lista de blocos.
Selecione um novo serviço vinculado e configure a sua conexão do Azure Synapse Analytics para se conectar ao banco de dados DWDB. Clique em Criar quando terminar.
Na configuração do conjunto de dados, selecione Criar tabela e insira o esquema Dbo e o nome da tabela Ratings. Clique em OK quando tiver concluído.
Como uma condição upsert foi especificada, você precisa acessar a guia Configurações e selecionar 'Permitir upsert' com base nas colunas de chave PrimaryGenre e year.
Neste ponto, você terminou de criar o seu Fluxo de Dados de Mapeamento com oito transformações. Está na hora de executar o pipeline e ver os resultados.
Tarefa 5: Executar o pipeline
Acesse a guia pipeline1 na tela. Como o Azure Synapse Analytics no Fluxo de Dados usa o PolyBase, você precisa especificar uma pasta de preparo do ADLS ou do blob. Na guia de configurações da atividade Execute Data Flow, abra a seção PolyBase, selecione o serviço vinculado do ADLS e especifique um caminho da pasta de preparo.
Antes de publicar o seu pipeline, realize outra execução de depuração para confirmar que ele está funcionando conforme o esperado. Observando a guia Saída, você pode monitorar o status de ambas as atividades enquanto elas estão em execução.
Depois que ambas as atividades forem bem-sucedidas, você poderá clicar no ícone de óculos ao lado da atividade Data Flow para obter uma visão mais detalhada da execução de Data Flow.
Se você tiver usado a mesma lógica descrita neste laboratório, o seu Fluxo de Dados gravará 737 linhas no DW do SQL. Você pode entrar no SQL Server Management Studio para verificar se o pipeline funcionou corretamente e ver o que foi gravado.
