Programa
Snowflake Foundations
0 min
Os pipelines de dados são um componente central da engenharia de dados. Eles representam a série de etapas que movem os dados brutos dos sistemas de origem para um warehouse de dados, onde eles podem ser limpos, transformados e preparados para análise. Um pipeline bem projetado garante que os dados fluam de forma eficiente, confiável e segura, desde a ingestão até o insight.
A Snowflake, uma poderosa plataforma de dados baseada em nuvem, é especialmente adequada para a criação de pipelines de dados escaláveis, flexíveis e econômicos. Com recursos incorporados para carregamento, transformação, automação e monitoramento de dados, o Snowflake permite que as equipes de dados simplifiquem fluxos de trabalho complexos e se concentrem em fornecer valor a partir dos dados.
Neste tutorial, você conhecerá o processo de criação de um pipeline de dados no Snowflake, desde a configuração do seu ambiente até a automatização das transformações de dados usando ferramentas nativas como Streams e Tasks. Para isso, você precisará conhecer o processo de integração de um serviço em nuvem e os sistemas de gerenciamento de eventos. Para isso, recomendo que você consulte a documentação de seu serviço de nuvem específico.
Se você está apenas começando a usar o Snowflake, considere nossa programa de habilidades Fundamentos do Snowflake para que você possa colocar a mão na massa e aprender a consultar, modelar e gerenciar dados.
Entendendo a arquitetura do Snowflake
A arquitetura exclusiva do Snowflake foi bem projetada para oferecer suporte a pipelines de dados. O Snowflake separa a computação e o armazenamento, o que significa que você pode dimensionar cada um independentemente.
Os dados são armazenados centralmente na camada de armazenamento em nuvem, enquanto a computação ocorre em warehouse virtuais, que podem ser iniciados, interrompidos e redimensionados conforme necessário.
Essa arquitetura permite que vários usuários e processos acessem os mesmos dados sem contenção, o que a torna ideal para cargas de trabalho de pipeline que envolvem ingestão, transformação e análise.
Ele também adiciona uma camada de segurança em que as alterações nos dados podem ocorrer sem afetar diretamente a capacidade do usuário de consultar esses dados.
Para obter mais informações e saber mais sobre a camada de nuvem que não foi discutida aqui, recomendo a você este um mergulho profundo na arquitetura do Snowflake.
Componentes-chave para pipelines Snowflake
Em cada pipeline, devemos considerar as etapas necessárias. Geralmente, você prepara os dados antes de armazená-los em um warehouse virtual. Esse warehouse virtual conterá o banco de dados que armazena as tabelas e seus esquemas. Aqui estão as definições desses elementos essenciais usados nos pipelines de dados do Snowflake:
- Estágios: Locais de armazenamento temporário ou externo usados para carregar/descarregar dados.
- Warehouses virtuais: Recursos de computação usados para processar consultas SQL e operações de dados.
- Bancos de dados: Recipientes lógicos para organizar seus dados. Geralmente, eles armazenam tabelas de coleta
- Esquemas: Subdivisões em um banco de dados para agrupar tabelas e objetos relacionados. Isso define como as tabelas estão relacionadas umas com as outras.
- Tabelas: Onde os dados estruturados são armazenados
Configurando seu ambiente Snowflake
Vamos garantir que você tenha um ambiente Snowflake pronto para funcionar. Se você é novo no Snowflake, siga estas etapas para começar:
Criar uma conta Snowflake
Se você não tiver uma conta Snowflake, há uma avaliação gratuita de 30 dias que inclui (no momento em que escrevo) US$ 400 em uso gratuito. Isso deve ser suficiente para que você possa testar alguns pipelines de amostra!
- Visite Teste do Snowflake e inscreva-se para obter uma conta gratuita.
Formulário de inscrição para a Prova de Snowflake
- Para nossos propósitos, selecionaremos a edição Standard em vez da edição Enterprise para economizar em custos de crédito no futuro se você continuar a usar o Snowflake!
- Escolha o provedor de nuvem de sua preferência (AWS, Azure ou GCP). Os exemplos deste tutorial usarão o AWS.
- O data center escolhido automaticamente deve ser o mais próximo de você geograficamente e deve oferecer a melhor experiência.
- Após a criação da conta, faça login na Snowflake Web UI.
Navegando na interface do usuário
Ao entrar, você verá:
- Planilha: Onde você pode executar consultas SQL.
- Bancos de dados: Visualizar e gerenciar bancos de dados e esquemas.
- Warehouses: Monitore e controle os recursos de computação.
- História: Programar consultas e tarefas executadas anteriormente
Se você quiser mais conselhos sobre o uso do Snowflake, siga este link Fundamentos do Snowflake curso.
Criar os recursos necessários
Nesta etapa, você verá como criar um warehouse virtual, um banco de dados e um esquema. Execute o seguinte SQL em sua planilha para configurar seu ambiente:
-- set the user role appropriately
USE ROLE accountadmin;
-- Create a virtual warehouse
CREATE WAREHOUSE datacamp_pipeline_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;
-- Set our active working warehouse
USE WAREHOUSE datacamp_pipeline_wh;
-- create our Datacamp Sample Database
CREATE OR REPLACE DATABASE datacamp_sample_db;
-- create the Datacamp (Sample) Raw Schema
CREATE OR REPLACE SCHEMA datacamp_sample_db.raw_schema;Carregando dados no Snowflake
Como discutimos em nosso artigo sobre Ingestão de dados no Snowflakevocê pode usar o Snowflake para carregar dados em lote (bulk) e continuamente. O carregamento em massa foi projetado para que você carregue grandes quantidades de dados na tabela de uma só vez. Isso geralmente é feito em intervalos programados ou automatizado com o gerenciamento baseado em eventos do Snowpipe.
O carregamento contínuo (fluxo) pode ser realizado com o uso de itens como o Snowpipe Stream. Para ingerir dados, precisamos criar integrações de dados (estágios) com nosso provedor de nuvem e carregar os dados usando COPY INTO.
Criação do estágio S3 e integração
A ingestão de dados é a primeira etapa de qualquer pipeline. Nesta etapa, abordaremos os conceitos básicos da criação de uma integração de armazenamento e da criação de um estágio. Para isso, devemos imaginar que você já conectou o Snowflake ao AWS com um usuário IAM do Snowflake. Caso contrário, consulte o Guia de configuração do Snowflake AWS na documentação.
-- set Role Context
USE ROLE accountadmin;
-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;
-- this creates the storage integration
-- The STORAGE_AWS_ROLE_ARN will come from the AWS configuration guide steps
-- We will allow access to all locations and block any with sensitive data
CREATE STORAGE INTEGRATION datacamp_aws
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = 'S3'
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = '<iam_role>'
STORAGE_ALLOWED_LOCATIONS = ('*')
STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);
-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format
TYPE = 'CSV'
FIELD_OPTIONALLY_ENCLOSED_BY = '"';
USE SCHEMA datacamp_sample_db.raw_schema;
--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
STORAGE_INTEGRATION = datacamp_aws
URL = 's3://bucket1/path1/'
FILE_FORMAT = my_csv_format;No final, você deverá ter criado com sucesso a integração do Snowflake com o AWS e criado o estágio. Se você quiser revisar os conceitos básicos da AWS, o artigo a seguir é uma ótima introdução à AWS.
Carregamento em massa com COPY INTO
O comando COPY INTO é o método comum para fazer upload de dados para o Snowflake. Você pode usar um arquivo carregado localmente no Snowflake ou usar nossas conexões de nuvem. No exemplo a seguir, usaremos nosso estágio para copiar um arquivo CSV.
-- Load data using COPY INTO
-- The following will attempt to load all file in the bucket
-- To specify the particular files you can use the FILES or PATTERN settings
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @my_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Automatizando o carregamento com o Snowpipe
O carregamento manual sempre que recebemos um novo arquivo é ineficiente e consome muito tempo à medida que a velocidade aumenta. Em vez disso, podemos usar o Snowpipe para carregar arquivos do nosso palco. Em seguida, combinamos nossas notificações de eventos do AWS com nosso usuário IAM do Snowflake para acionar o Snowpipe todas as vezes.
-- Let’ s first create our Snowpipe
CREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Em seguida, podemos usar algo como Amazon SQS para enviar notificações ao nosso usuário IAM sempre que o bucket de destino receber um arquivo.
Por fim, podemos monitorar nosso Snowpipe usando:
SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;Agora você não precisa mais se preocupar em fazer upload de dados sempre que um arquivo chegar ao seu bucket do S3! Essas notificações na nuvem acionarão o Snowpipe para que você carregue os dados.
Transformação de dados com fluxos e tarefas
Os programas e as tarefas do Snowflake são ferramentas integradas para que você acompanhe e transforme dados em constante mudança. Os programas se concentram no rastreamento de alterações em tabelas em que as tarefas são utilizadas para automatizar a execução de instruções SQL.
Introdução aos fluxos
Um fluxo captura informações de captura de dados de alteração (CDC) - ou seja, inserções, atualizações e exclusões - em uma tabela desde a última vez em que ela foi consultada. Geralmente, essas consultas são feitas na forma de DML (Data Manipulation Language, linguagem de manipulação de dados). Com isso, você pode ver as alterações em uma tabela entre dois tempos de consulta diferentes.
-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Você pode então consultar o fluxo para obter apenas as alterações:
SELECT * FROM my_stream WHERE METADATA$ACTION = 'INSERT';Introdução às tarefas
As tarefas permitem que você automatize a execução de SQL em um cronograma ou acionador de evento. As tarefas podem recorrer a SQL, Javascript, Python e assim por diante, para que você automatize procedimentos em nossas tabelas e pipelines.
-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
WAREHOUSE = datacamp_sample_db
SCHEDULE = '5 MINUTE'
AS
INSERT INTO raw_schema.aws_processed_data
SELECT * FROM my_stream WHERE location = ‘USA’;
-- This will allow us to test the task
EXECUTE TASK process_changes_task;Combinação de fluxos e tarefas
Essa é a base para os pipelines de transformação automatizados. Como você viu acima, as tarefas podem usar Streams para extrair novos dados para uma tabela diferente. Um processo geral pode ser parecido com o seguinte:
- Criar um fluxo para que você possa monitorar as alterações em uma tabela de dados
- Crie uma tarefa que extraia dados desse fluxo em uma base programada
- Com cada extração de dados, transforme os dados no formato desejado
- Carregue esses dados em nossa tabela de destino
Dessa forma, criamos partes de um pipeline de dados. Combinando Streams e Tasks, podemos realizar transformações somente nos dados recém-carregados. Com apenas tarefas, teríamos que processar todas as nossas tabelas brutas sempre que quiséssemos novos dados, o que pode ser caro e ineficiente no uso do poder de computação.
Criação de um pipeline de dados de ponta a ponta
Vamos examinar um caso de uso simples de pipeline: ingestão de dados de pedidos, rastreamento de alterações e transformação para análise. Como abordamos como executar cada etapa de forma independente, esta seção aborda como as etapas podem ser vistas em conjunto.
Etapa 1: Criar integração e estágio de armazenamento em nuvem
Primeiro, temos que conectar nosso ecossistema Snowflake ao nosso armazenamento em nuvem. Usamos o AWS como exemplo. No Snowflake, usamos os comandos CREATE STORAGE INTEGRATION e CREATE STAGE para criar a conexão entre o AWS e o nosso banco de dados.
-- set Role Context
USE ROLE accountadmin;
-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;
-- this creates the storage integration
CREATE STORAGE INTEGRATION datacamp_aws
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = 'S3'
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = '<iam_role>'
STORAGE_ALLOWED_LOCATIONS = ('*')
STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);
-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format
TYPE = 'CSV'
FIELD_OPTIONALLY_ENCLOSED_BY = '"';
USE SCHEMA datacamp_sample_db.raw_schema;
--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
STORAGE_INTEGRATION = datacamp_aws
URL = 's3://bucket1/path1/'
FILE_FORMAT = my_csv_format;Etapa 2: Ingerir dados com o Snowpipe
Nesta segunda etapa, criamos um Snowpipe que executará o comando COPY INTO em nossas tabelas de dados brutos.
-- Let’ s first create our SnowpipeCREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Etapa 3: Use Streams para capturar alterações
Em seguida, usamos um fluxo nessa tabela de dados brutos para que você possa acompanhar as alterações.
-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Etapa 4: Automatize as transformações com Tasks
Em seguida, criamos uma tarefa que transforma e carrega esses dados recém-carregados em nossa versão processada da tabela. Isso será acionado em um cronograma para verificar se há novos dados. Se não houver dados novos, a tarefa não será executada.
-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
WAREHOUSE = datacamp_sample_db
SCHEDULE = '5 MINUTE'
AS
INSERT INTO raw_schema.aws_processed_data
SELECT * FROM my_stream WHERE location = ‘USA’;
-- This will allow us to test the task
EXECUTE TASK process_changes_task;Etapa 5: Monitorar e gerenciar
Por fim, queremos ter certeza de que nossos pipelines estão funcionando como esperado. Certifique-se de usar os comandos SHOW PIPES e SHOW TASKS para verificar se eles estão configurados e funcionando corretamente. Em seguida, use as tabelas de metadados do site INFORMATION_SCHEMA para acompanhar o desempenho e os erros de carregamento de dados.
Práticas recomendadas para o desenvolvimento de pipeline de dados no Snowflake
Como sempre, queremos garantir que estamos operando com as práticas recomendadas em nosso desenvolvimento. O que você deve considerar é a otimização do desempenho e do custo, garantindo a qualidade e a confiabilidade dos dados e implementando os níveis adequados de segurança e acesso.
Otimize o desempenho e o custo
Nossa meta deve ser sempre operar nossos dutos da forma mais eficiente possível e com o menor custo possível. Dessa forma, estamos minimizando nosso impacto nos servidores e fornecendo dados da maneira mais oportuna possível. Aqui estão algumas dicas:
- Use a suspensão automática nos warehouses para evitar encargos de inatividade. Isso reduz os custos de computação com a operação dos warehouses, já que eles cobram por tempo.
- Carregue arquivos compactados e agrupe-os em tamanhos ideais. Arquivos muito pequenos consomem muita sobrecarga da rede, pois você precisa gerar conexões para carregar cada arquivo. Arquivos muito grandes podem consumir muita largura de banda e correm o risco de falhar. A Snowflake recomenda deltas de 100 a 250 MB de tamanho.
- Programe as transformações durante as janelas de baixa utilização. Isso significa que você não está competindo por recursos.
Garantir a qualidade e a confiabilidade dos dados
A qualidade e a confiabilidade dos dados são essenciais para todos os engenheiros de dados. Você quer ter certeza de que os dados que estão sendo carregados podem ser usados sem erros. Você também quer ter certeza de que seu pipeline de dados não fará com que os dados caiam inadvertidamente quando houver problemas. Aqui estão algumas dicas para o gerenciamento da qualidade dos dados:
- Valide os dados antes e depois da ingestão. Isso pode ser feito com verificações periódicas das tabelas de preparação e alertas.
- Use fluxos e tabelas de preparação para isolar e corrigir registros problemáticos. Como somente os dados alterados são extraídos, podemos minimizar o escopo de nossos problemas usando fluxos.
- Aplique restrições exclusivas ou lógica de deduplicação quando necessário. Isso pode evitar coisas como
NULLdados onde não podemos ter nenhum ou pré-deduplicação no caso de extração de dados sobrepostos.
Implemente controles de segurança e acesso
Por fim, é importante garantir o acesso e a segurança adequados aos seus dados. Algumas tabelas podem conter informações confidenciais ou não devem ser alteradas, exceto por um grupo seleto. Siga este conselho para gerenciar adequadamente a segurança e o acesso que você tem no Snowflake:
- Uso funções e concessões para restringir o acesso.
- Proteja dados confidenciais com mascaramento em nível de coluna ou políticas de acesso por linha.
- Faça a rotação das credenciais de armazenamento e monitore o uso do palco.
Conclusão
Ao criar pipelines de dados no Snowflake, você tem um processo simplificado e escalável, perfeito para iniciantes no mundo da engenharia de dados. Com sua arquitetura intuitiva, opções robustas de ingestão e ferramentas de automação integradas, como Streams e Tasks, o Snowflake capacita as equipes de dados a passar rapidamente de dados brutos para insights acionáveis.
Seguindo este guia, você agora tem o conhecimento básico para configurar seu próprio pipeline Snowflake, lidando com ingestão, transformação e orquestração, tudo em uma única plataforma. Para obter mais informações sobre o Snowflake, confira os seguintes guias:
