O que é Prefeito?

É uma estrutura flexível para construir, executar de forma confiável e observar seu fluxo de dados enquanto oferece suporte a vários padrões de execução e acesso a dados. Ele permite que você transforme qualquer script Python em um aplicativo totalmente operacionalizado .

O que é o Fivetran?

É uma plataforma de integração de dados que permite às empresas replicar dados de vários sistemas em um data warehouse central. Ele oferece conectores pré-construídos para uma ampla variedade de fontes de dados, incluindo bancos de dados, aplicativos SaaS, armazenamento em nuvem, logs de eventos e muito mais. Fivetran pode ajudá-lo a manter seus dados consistentes, completos e atualizados automatizando o processo de integração de dados.

Por que Fivetran + Prefeito?

Prefect coordena o fluxo de dados em toda a sua pilha. Com Fivetran, você pode facilmente mover dados de fontes para destinos. Prefeito permite:

• agende e observe suas sincronizações de replicação de dados,
• adicione uma sincronização à parte do pipeline que precisa operar com os dados mais atualizados de qualquer sistema de origem,
• garanta um fluxo de dados confiável por meio de novas tentativas, alertas e gerenciamento de dependências.

Primeiro, inscreva-se em uma conta gratuita do Prefect Cloud e crie um espaço de trabalho . Em seguida, instale o Prefect e faça login em seu espaço de trabalho na nuvem a partir de um terminal:

pip install prefect
prefect cloud login

Em seguida, crie um script chamado myflow.py:

from prefect import flow

@flow(log_prints=True)
def hello():
    print("You're the Prefectionist now!")

if __name__ == "__main__":
    hello()

Introdução ao Fivetran

Para começar a usar o Fivetran, inscreva-se em uma conta de avaliação . Em seguida, uma tela de boas-vindas o guiará pela configuração inicial para configurar sua primeira sincronização Fivetran selecionando uma fonte, destino e dados a serem integrados.

Conectar fonte

Para começar, você pode selecionar o GitHub como um conector e escolher os repositórios para sincronizar. Isso permitirá que você execute consultas SQL para analisar suas estrelas do GitHub, solicitações pull, etc.

Conectar destino

Você pode selecionar um destino a partir de uma variedade de opções. A maioria deles são armazéns de dados em nuvem e bancos de dados em nuvem:

Para esta demonstração, escolheremos Snowflake. Depois de selecionar o destino, você obtém um trecho de transação SQL que pode ser executado em sua planilha Snowflake para configurar tudo. Assim que executar esse script, você pode definir os campos correspondentes ( nome de usuário, senha, banco de dados, função ) no formulário à esquerda:

O mesmo bloco de código está disponível na documentação do Fivetran . Se o teste de conexão passar, você pode continuar com a seleção de dados:

Selecionar dados

Algumas colunas podem ser bloqueadas ou com hash para privacidade e conformidade:

Para esta demonstração, selecione: “Sincronizar todos os dados” e clique em “Continuar”.

Novo conector: Planilhas Google

Outro conector simples pode replicar dados de uma planilha do Google . Você pode criar um que se pareça com a seguinte tabela :

Depois de criar uma tabela, vá para Dados → Intervalos nomeados :

Selecione as três primeiras colunas e salve-as como um intervalo nomeado, por exemplo Fivetran. Por exemplo, para definir todas as linhas das colunas A, B e C em Sheet1, você pode usar Sheet1!A:C. Isso é importante para que o Fivetran saiba quais linhas e colunas você deseja sincronizar.

No Fivetran, cole o URL da planilha e selecione o intervalo nomeado. Em seguida, salve e teste o conector:

Se tudo funcionar conforme o esperado, você pode selecionar “Sincronizar todos os dados” e “Continuar”:

Agora você pode iniciar a sincronização inicial:

Após a sincronização, você verá um novo esquema e uma tabela correspondente a essa planilha do Google em sua planilha do Snowflake.

Automatize a replicação de dados com Prefect e Fivetran

Gerar uma chave de API Fivetran

Para agendar e orquestrar suas sincronizações, seus fluxos Prefect precisam ser autenticados com o Fivetran. Para fazer isso, você precisa de uma chave de API. Vá para as configurações da sua conta e gere uma chave de API conforme mostrado abaixo:

Isso fornecerá um identificador de chave de API e um token secreto de API :

Crie um Bloco Prefeito para Fivetran

Instale a prefect-fivetran coleção e registre um bloco Fivetran:

pip install prefect prefect_fivetran
prefect block register -m prefect_fivetran

E aqui está como você pode fazer isso com código:

from dotenv import load_dotenv
import os
from prefect_fivetran import FivetranCredentials

load_dotenv()

fivetran_credentials = FivetranCredentials(
    api_key=os.environ.get("FIVETRAN_API_KEY"),
    api_secret=os.environ.get("FIVETRAN_API_SECRET_KEY"),
)
fivetran_credentials.save("default")

Por fim, podemos executar um fluxo gsheet.pypara acionar uma sincronização:

from prefect import flow
from prefect_fivetran import FivetranCredentials
from prefect_fivetran.connectors import (
    wait_for_fivetran_connector_sync,
    start_fivetran_connector_sync,
)


@flow
def example_flow(connector_id: str):
    fivetran_credentials = FivetranCredentials.load("default")

    last_sync = start_fivetran_connector_sync(
        connector_id=connector_id,
        fivetran_credentials=fivetran_credentials,
    )

    return wait_for_fivetran_connector_sync(
        connector_id=connector_id,
        fivetran_credentials=fivetran_credentials,
        previous_completed_at=last_sync,
        poll_status_every_n_seconds=60,
    )


if __name__ == "__main__":
    example_flow("bereft_indices")

Acionando uma sincronização de replicação de dados

Para validar se a integração Fivetran, Prefect e Snowflake está configurada corretamente, acionaremos esse fluxo duas vezes. Primeiro, usaremos a tabela do Google Sheet como está. Em seguida, modificaremos alguns registros e inspecionaremos se o fluxo de trabalho de sincronização os atualizou no destino. Dessa forma, podemos validar se a tabela Snowflake permanece sincronizada com o sistema de origem.

Vamos acionar esse fluxo pela primeira vez:

python gsheet.py

      
                

                    
                    
                
            

A sincronização foi bem-sucedida e nada foi alterado na tabela Snowflake como resultado da primeira execução do fluxo:

Vamos agora alterar o nome do primeiro cliente de Michael para Mike, remover o registro do cliente Shawn com ID 2 e adicionar outro cliente com ID 101:

Depois de fazer essas alterações, acione o fluxo do Prefeito novamente:

python gsheet.py

Podemos validar que os dados modificados foram replicados corretamente para a tabela Snowflake:

Dimensões que mudam lentamente

Observe que as alterações que refletem a atualização do primeiro nome de Michael para Mike e a exclusão do registro de Shawn não são visíveis nesta tabela. Se você estiver interessado em rastrear essas informações como parte de sua estratégia de Dimensões de Alteração Lenta , precisará habilitar o Histórico para essa sincronização. Para obter uma lista de conectores que oferecem suporte ao histórico de rastreamento, verifique a documentação do Fivetran do modo de histórico .

Agendando suas sincronizações

Se você deseja executar este script dentro do cronograma, por exemplo, todos os dias às 9h, pode usar o seguinte comando para criar uma implantação do Prefect :

prefect deployment build --cron "0 9 * * *" gsheet.py:example_flow -n dev -a

prefect agent start -q default

Próximos passos

Esta postagem demonstrou como começar com Prefect e Fivetran e criar suas primeiras sincronizações de replicação de dados do GitHub e Google Sheets para Snowflake. Vimos como você pode armazenar chaves de API com segurança usando Prefect Blocks e orquestrar sincronizações Fivetran a partir de um fluxo Prefect.

Se algo discutido neste post não estiver claro, sinta-se à vontade para me marcar ao fazer uma pergunta no Prefect Community Slack ou no Prefect Discourse .

Obrigado pela leitura e boa engenharia!