Como Transmitir Dados da Web Para o Snowflake com Scrapeless e Streaming do Snowpipe

Principais Conclusões:

• Dados da web raspados em Snowflake sem um esquema fixo. O Scrapeless Scraping Browser renderiza uma página em um navegador na nuvem e emite JSON delimitado por novas linhas (NDJSON); o Snowflake o ingere em uma coluna VARIANT, então novos campos nunca interrompem o carregamento.
• Quatro métodos de ingestão, uma forma de dados. Bulk COPY INTO para carregamentos únicos, Snowpipe para lotes contínuos baseados em arquivos, Snowpipe Streaming para linhas de baixa latência, e o conector Kafka para pipelines baseados em eventos — todos leem o mesmo NDJSON produzido pelo Scrapeless.
• A arquitetura de alto desempenho do Snowpipe Streaming está geralmente disponível (GA desde setembro de 2025). Ele escreve linhas diretamente através de um SDK (Java, Python, Node.js) ou REST, com canais, tokens de deslocamento e recuperação exatamente uma vez — sem arquivos em estágio.
• Schema-on-read mantém os dados raspados flexíveis. Consulte uma coluna VARIANT com a notação col:field::type e desagregue matrizes com LATERAL FLATTEN — nenhuma migração quando a página de origem adiciona um campo.
• A CLI do Snowflake (snow) é a ferramenta atual. pip install snowflake-cli, então snow sql -f ingest.sql executa toda a configuração a partir de um arquivo.
• Gratuito para começar. Novas contas Scrapeless incluem tempo de execução gratuito do Scraping Browser — inscreva-se em Scrapeless Website.

Introdução: da página renderizada para uma tabela Snowflake

As equipes de análise estão cada vez mais interessadas em dados da web — catálogos de produtos, listagens, avaliações, sinais de mercado — no mesmo armazém que seus dados de primeira parte, para que possam se unir, modelar e alimentar BI. O Snowflake é um destino comum porque seu tipo VARIANT armazena JSON semiestruturado nativamente e o torna consultável com SQL.

A fricção está na lacuna entre os dois sistemas. As páginas raspadas são renderizadas em JavaScript e estão por trás de defesas contra bots; os dados frequentemente chegam como JSON aninhado cuja forma muda à medida que o site de origem altera. Carregadores feitos à mão que mapeiam cada campo para uma coluna quebram na primeira vez que a página adiciona um.

Este post apresenta um fluxo de trabalho iniciado no terminal que fecha essa lacuna. O Scrapeless Scraping Browser cuida da renderização e do lado anti-detecção e emite NDJSON; o Snowflake o ingere de quatro maneiras diferentes, dependendo de quão frescos os dados precisam ser. O produtor de exemplo é o ambiente público de raspagem books.toscrape.com, então cada comando abaixo é reproduzível — o mesmo padrão se aplica a alvos mais difíceis (veja os guias irmãos Melhores Raspadores da Zillow em 2026 e Melhores Raspadores da Amazon em 2026).

O que você pode fazer com isso

• Construa um lakehouse de dados da web. Lande catálogos e listagens raspados no Snowflake e junte-os a dados internos de vendas ou inventário.
• Execute instantâneas de mercado programadas. Envie um novo arquivo NDJSON por execução para um estágio e deixe o Snowpipe carregá-lo automaticamente em minutos.
• Alimente painéis de controle quase em tempo real. Transmita eventos raspados linha por linha com o Snowpipe Streaming para frescor de menos de um minuto.
• Conecte-se a uma infraestrutura Kafka existente. Envie registros raspados para um tópico e deixe o conector Kafka do Snowflake os aterrissar.
• Mantenha o esquema flexível. Armazene o JSON bruto em VARIANT e molde-o no momento da consulta, assim alterações no site de origem nunca bloqueiam um carregamento.

No Scrapeless, acessamos apenas dados públicos disponíveis, enquanto respeitamos rigorosamente as leis aplicáveis, regulamentos e políticas de privacidade dos sites. O conteúdo deste post é apenas para fins de demonstração.

Por que o Scrapeless Scraping Browser

O Scrapeless Scraping Browser é um navegador em nuvem personalizável e anti-detecção, projetado para rastreadores da web e agentes de IA. Como o lado produtor de um pipeline do Snowflake, ele oferece:

• Renderização JavaScript do lado da nuvem, para que os dados estejam presentes no DOM antes da extração
• Proxies residenciais em mais de 195 países, ancorados por sessão
• Fingerprinting de navegador anti-detecção
• Uma única superfície CLI scrapeless-scraping-browser cuja eval retorna JSON que você pode remodelar em NDJSON em um único passo
• Persistência de sessão para raspagens de várias páginas

Obtenha sua chave API no plano gratuito em Scrapeless Website.

Pré-requisitos

• Node.js 18 ou superior
• Uma conta Scrapeless e chave API — inscreva-se em Scrapeless Website
• Uma conta Snowflake com um papel que possa criar bancos de dados, armazéns, estágios e pipes
• A CLI do Snowflake: pip install snowflake-cli (Python 3.10+)
• Para ingestão automática do Snowpipe e estágios externos: um bucket na nuvem (AWS S3, GCS ou Azure) e permissão para criar uma integração de armazenamento
• jq é opcional (um fallback de uma linha do Node é mostrado para a conversão de NDJSON)

Configure uma conexão Snowflake uma vez, em ~/.snowflake/config.toml:

[connections.demo]
account   = "myorg-myaccount"
user      = "jondoe"

senha = "sua_senha_aqui"
armazém = "ingest_wh"
banco_de_dados = "web_data"
esquema = "raw"
papel = "sysadmin"

Então `snow sql -c demo -q "SELECT CURRENT_VERSION();"` confirma que funciona.

---

## O pipeline em um relance

Scrapeless Scraping Browser → arquivo NDJSON → estágio Snowflake → tabela (VARIANT)
(renderizar + extrair) (um objeto (interna ou COPY INTO | única vez
por linha) externa) Snowpipe | contínuo
Streaming | baixa latência
Kafka | orientado a eventos

A estrutura nunca muda: Scrapeless emite um objeto JSON por linha, o arquivo é colocado em um estágio, e um dos quatro métodos o carrega em uma coluna `VARIANT` que você consulta com SQL.

---

## Passo 1 — Produzir NDJSON com Scrapeless

Instale o CLI e defina sua chave:

```bash
npm install -g scrapeless-scraping-browser
scrapeless-scraping-browser config set apiKey seu_token_api_aqui

Abra uma sessão na nuvem, navegue até a página do catálogo, aguarde um marcador estável e extraia os registros dos livros com eval. O JSON new-session aninha o id sob data.taskId — use jq, ou a alternativa portátil de grep mostrada:

# abrir uma sessão e capturar o id da tarefa (o caminho jq é .data.taskId)
SID=$(scrapeless-scraping-browser new-session --name books --ttl 300 --proxy-country US --json | jq -r '.data.taskId')
# sem jq? alternativa portátil:
# SID=$(scrapeless-scraping-browser new-session --name books --ttl 300 --proxy-country US --json | grep -oE '"taskId":"[^"]*"' | head -1 | cut -d'"' -f4)

# renderizar a página do catálogo, então aguardar pela grade de produtos
scrapeless-scraping-browser --session-id "$SID" open "https://books.toscrape.com/catalogue/page-1.html"
scrapeless-scraping-browser --session-id "$SID" wait "article.product_pod"

# extrair um registro por livro; o eval retorna um array JSON
scrapeless-scraping-browser --session-id "$SID" eval '
  JSON.stringify(Array.from(document.querySelectorAll("article.product_pod")).map(el => ({
    title: el.querySelector("h3 a")?.getAttribute("title") ?? null,
    price: el.querySelector(".price_color")?.textContent.trim() ?? null,
    rating: el.querySelector("p.star-rating")?.className.replace("star-rating", "").trim() ?? null,
    in_stock: /In stock/i.test(el.querySelector(".availability")?.textContent ?? ""),
    url: el.querySelector("h3 a")?.href ?? null
  })))
' > books.raw.json

scrapeless-scraping-browser --session-id "$SID" close

Converta o array para NDJSON — um objeto por linha, que é o formato que os carregadores do Snowflake leem com mais clareza:

# com jq
jq -c '.[]' books.raw.json > books.ndjson

# ou, sem jq, uma linha de comando em Node
node -e 'JSON.parse(require("fs").readFileSync("books.raw.json","utf8")).forEach(o=>console.log(JSON.stringify(o)))' > books.ndjson

books.ndjson agora contém um objeto JSON autossuficiente por linha. Se uma sessão fria retornar uma concha vazia ou um os error 10054 transitório, feche a sessão, crie uma nova e tente novamente um número limitado de vezes antes de extrair.

Passo 2 — Preparar Snowflake

Crie o armazém, banco de dados, esquema, um formato de arquivo JSON e uma tabela de aterrissagem com uma única coluna VARIANT. Salve isso como setup.sql e execute com snow sql -c demo -f setup.sql:

CREATE WAREHOUSE IF NOT EXISTS ingest_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;
CREATE DATABASE  IF NOT EXISTS web_data;
CREATE SCHEMA    IF NOT EXISTS web_data.raw;

USE WAREHOUSE ingest_wh;
USE SCHEMA web_data.raw;

-- NDJSON: um objeto JSON por linha, portanto não remova um array externo
CREATE OR REPLACE FILE FORMAT ndjson_format
  TYPE = JSON
  STRIP_OUTER_ARRAY = FALSE
  COMPRESSION = AUTO;

-- aterrissar o registro bruto como está; modelá-lo no momento da consulta
CREATE OR REPLACE TABLE raw_books (
  src       VARIANT,
  loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

STRIP_OUTER_ARRAY = FALSE está correto para NDJSON porque cada linha já é seu próprio objeto — STRIP_OUTER_ARRAY = TRUE é apenas para um arquivo que é um grande array [ ... ].

Passo 3 — Método 1: Carga em massa única com COPY INTO

Para um único arquivo ou um lote manual, coloque o arquivo no estágio e execute COPY INTO. O caminho mais simples é um estágio interno nomeado mais PUT:

-- um estágio interno nomeado vinculado ao formato JSON
CREATE OR REPLACE STAGE books_stage FILE_FORMAT = ndjson_format;

# enviar o NDJSON local para o estágio interno (o CLI do snow executa PUT)
snow sql -c demo -q "PUT file://$(pwd)/books.ndjson @books_stage AUTO_COMPRESS=TRUE OVERWRITE=TRUE"

-- carregar cada objeto como uma linha na coluna VARIANT
COPY INTO raw_books (src)
  FROM @books_stage
  FILE_FORMAT = (FORMAT_NAME = 'ndjson_format')
  ON_ERROR = 'CONTINUE';

Para mapear chaves JSON diretamente para colunas tipadas em vez de um VARIANT, crie uma tabela cujos nomes das colunas correspondam às chaves e use MATCH_BY_COLUMN_NAME:

CREATE OR REPLACE TABLE livros (
  titulo VARCHAR, preco VARCHAR, avaliacao VARCHAR, em_estoque BOOLEAN, url VARCHAR
);

COPY INTO livros
  FROM @livros_stage
  FILE_FORMAT = (TYPE = 'JSON')
  MATCH_BY_COLUMN_NAME = 'CASE_INSENSITIVE';

Se preferir deixar o Snowflake derivar o esquema a partir dos arquivos em estágio, INFER_SCHEMA com CREATE TABLE … USING TEMPLATE constrói a lista de colunas para você:

CREATE OR REPLACE TABLE livros_auto
  USING TEMPLATE (
    SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
    FROM TABLE(INFER_SCHEMA(
      LOCATION => '@livros_stage',
      FILE_FORMAT => 'ndjson_format'
    ))
  );

Para dados que já estão em um bucket na nuvem, aponte para um estágio externo em vez de fazer upload. Com uma integração de armazenamento (sem chaves inline):

CREATE OR REPLACE STAGE livros_s3_stage
  URL = 's3://meu-bucket/raspados/livros/'
  STORAGE_INTEGRATION = minha_integracao_s3
  FILE_FORMAT = ndjson_format;

COPY INTO livros_brutos (src) FROM @livros_s3_stage;

Obtenha sua chave API no plano gratuito: app.scrapeless.com

Etapa 4 — Método 2: Lotes contínuos com Snowpipe

Quando o scraper coloca um novo arquivo em um bucket em um cronograma, o Snowpipe carrega cada arquivo automaticamente — sem COPY manual e sem armazém dedicado. Um pipe envolve uma instrução COPY INTO; com AUTO_INGEST = TRUE, uma notificação de evento em nuvem aciona o carregamento:

CREATE OR REPLACE PIPE livros_pipe
  AUTO_INGEST = TRUE
  AWS_SNS_TOPIC = 'arn:aws:sns:us-east-1:123456789012:bucket-raspado'
  AS
  COPY INTO livros_brutos (src)
  FROM @livros_s3_stage
  FILE_FORMAT = (TYPE = 'JSON');

No S3, o evento flui através do SNS/SQS para uma fila gerenciada pelo Snowflake; GCS usa Pub/Sub e Azure usa Event Grid, cada um conectado com uma integração de notificação. Se preferir chamar o Snowpipe explicitamente, deixe AUTO_INGEST não definido e envie os caminhos dos arquivos em estágio para o endpoint REST insertFiles, e depois consulte insertReport.

Duas observações operacionais da orientação do Snowflake:

• A cobrança é sem servidor e agora é cobrada com base no por-GB para os dados que o Snowpipe ingere — não há armazém para dimensionar, e o antigo componente por-arquivo foi aposentado.
• O tamanho do arquivo importa. Tente manter arquivos em torno de 100–250 MB compactados, e não estagie mais de cerca de uma vez por minuto; estagiar com mais frequência adiciona sobrecarga de gerenciamento de fila sem reduzir a latência. Agrupe pequenos lotes de raspagem em arquivos maiores antes do estagiamento.

O Snowpipe torna os dados disponíveis em minutos, o que se encaixa bem em instantâneas de mercado programadas.

Etapa 5 — Método 3: Linhas de baixa latência com Snowpipe Streaming

Quando a frescura precisa ser de segundos, e não minutos, o Snowpipe Streaming escreve linhas diretamente em uma tabela — sem arquivos em estágio. A arquitetura de alto desempenho está geralmente disponível desde setembro de 2025, com SDKs para Java, Python e Node.js, além de uma API REST sobre um núcleo de cliente compartilhado; o clássico Snowpipe Streaming baseado em arquivos está em um caminho de descontinuação.

O modelo tem três conceitos fundamentais:

• Canais — uma conexão de streaming nomeada e de longa duração com uma tabela. As linhas são comprometidas na ordem dentro de um canal.
• Tokens de offset — uma string que sua aplicação anexa a cada lote. Após um reinício, getLatestCommittedOffsetToken() informa a última posição confirmada de forma durável, para que você reproduza apenas o que segue — a base para entrega exatamente uma vez.
• Cobrança por throughput — créditos por GB descompactado ingerido, em vez de por arquivo.

A forma do cliente Java é pequena:

SnowflakeStreamingIngestClient client =
    SnowflakeStreamingIngestClientFactory.builder("CLIENTE_LIVROS")
        .setProperties(props).build();

OpenChannelRequest request = OpenChannelRequest.builder("CANAL_LIVROS")
    .setDBName("DADOS_WEB").setSchemaName("BRUTO").setTableName("LIVROS_BRUTOS")
    .setOnErrorOption(OpenChannelRequest.OnErrorOption.CONTINUE)
    .build();

SnowflakeStreamingIngestChannel channel = client.openChannel(request);
channel.insertRow(rowAsMap, offsetToken);   // um registro raspado
channel.getLatestCommittedOffsetToken();     // para recuperação

Utilize Streaming quando os registros raspados chegam em um fluxo contínuo (um agente emitindo eventos enquanto rasteja) e o painel precisa deles em segundos.

Etapa 6 — Método 4: Carregamentos baseados em eventos com o conector Kafka

Se os registros raspados já fluem pelo Apache Kafka, o Conector Snowflake para Kafka coloca um tópico em uma tabela (um tópico mapeia para uma tabela). Ele é executado dentro de um trabalhador Kafka Connect. Uma única propriedade seleciona o mecanismo de ingestão subjacente:

name=livros-raspados-sink
connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
topics=livros_raspados
snowflake.database.name=DADOS_WEB
snowflake.schema.name=BRUTO
# SNOWPIPE (baseado em arquivo, padrão) ou SNOWPIPE_STREAMING (baixa latência)
snowflake.ingestion.method=SNOWPIPE_STREAMING

Cada tabela que o conector cria possui duas colunas VARIANT: RECORD_CONTENT (o payload da mensagem) e RECORD_METADATA (tópico, partição, offset, timestamps e cabeçalhos). Consulte o payload exatamente como faria com qualquer VARIANT.

Para uma alternativa totalmente gerenciada — sem cluster Kafka Connect para operar — o Snowflake Openflow (disponível em geral, construído sobre o Apache NiFi) ingere dados de fontes Kafka, Kinesis, bancos de dados e SaaS para o Snowflake por meio de pipelines gerenciados.

Escolhendo um método

A maioria das equipes começa com COPY INTO para validar o esquema, migra para Snowpipe assim que o raspador é executado em uma programação, e adota Streaming ou Kafka apenas quando a frescura sub-minuto justifica a peça adicional em movimento.

Consultando os dados carregados

Como o registro bruto está em um VARIANT, você o molda no momento da leitura. Navegue com o operador : e converta com :::

SELECT
  src:title::string   AS title,
  src:price::string   AS price,
  src:rating::string  AS rating,
  src:in_stock::boolean AS in_stock,
  src:url::string     AS url
FROM raw_books;

Quando um registro raspado contém um array — uma lista de fotos, uma série de preços históricos — LATERAL FLATTEN o descompacta em linhas:

-- raw_listings: uma tabela hipotética de listagens raspadas carregada da mesma forma que raw_books.
-- (O exemplo de livros não tem array aninhado; isso mostra o padrão para uma fonte que tem.)
SELECT
  src:title::string AS title,
  ph.value:date::string  AS price_date,
  ph.value:price::number AS price
FROM raw_listings,
LATERAL FLATTEN(INPUT => src:priceHistory) ph;

Não é necessária migração quando a página de origem adiciona um campo — ele simplesmente aparece sob src na próxima carga.

O que você recebe de volta

Após COPY INTO, cada objeto raspado é uma linha em raw_books. A coluna VARIANT mantém o registro verbatim; o esquema abaixo é normativo e os valores dos campos são amostras ilustrativas:

// Um VARIANT src de uma linha, emitido pelo extrator da Etapa 1.
{
  "title": "A Light in the Attic",
  "price": "£51.77",
  "rating": "Three",
  "in_stock": true,
  "url": "https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html"
}

Algumas observações honestas:

• Os preços chegam como strings exibidas. Converta e limpe em SQL (REPLACE(src:price::string, '£', '')::number) em vez de esperar números da página.
• Campos condicionais são anuláveis. Um campo ausente em uma página específica simplesmente está faltando em src; o acesso a VARIANT retorna NULL em vez de erro.
• MATCH_BY_COLUMN_NAME ignora chaves não correspondentes. Novas chaves chegam a uma tabela VARIANT automaticamente, mas são descartadas por uma tabela tipada até que você adicione a coluna.
• O histórico de carga do Snowpipe é mantido por 14 dias nos metadados do pipe; o histórico de COPY em massa é mantido por 64 dias nos metadados da tabela — tenha isso em mente ao auditar uma complementação.

Conclusão

Colocar dados da web raspados no Snowflake se resume a quatro movimentos: renderizar e extrair com Scrapeless, emitir NDJSON, colocar o arquivo em estágio e carregá-lo com o método cuja latência se encaixa — COPY INTO, Snowpipe, Snowpipe Streaming ou o conector Kafka. Coloque o registro bruto em uma coluna VARIANT e molde-o no momento da consulta, para que o pipeline sobrevive à adição de campos pela página de origem.

Prenda a saída dos EUA quando o destino precisar, mantenha a cadeia de sessão do Scrapeless dentro de uma única invocação de shell, siga o padrão descobrir → extrair e trate os campos ausentes como anuláveis. Para alvos mais difíceis do que o sandbox usado aqui, o mesmo padrão de produtor se aplica — veja o guia irmão Melhores Scrapers da Zillow em 2026, a página do produto Scraping Browser e a documentação da Scrapeless.

Pronto para construir seu pipeline de dados alimentado por IA?

Junte-se à nossa comunidade para reclamar um plano gratuito e conectar-se com desenvolvedores que constroem pipelines de dados da web para o armazém: Discord · Telegram.

Inscreva-se no Site da Scrapeless para um tempo de execução gratuito do Scraping Browser, e veja scrapeless.com/en/pricing para escalar minutos de sessão e concorrência à medida que o pipeline cresce.

FAQ

Q1: É legal a raspagem da web para ingestão em armazém?
Coletar dados visíveis publicamente é amplamente defensável, mas a legalidade depende dos termos do site-alvo, da jurisdição e do tipo de dado. Revise os termos de serviço do site, evite dados pessoais ou restritos e consulte um advogado antes do uso comercial. O site sandbox usado aqui, books.toscrape.com, existe especificamente para prática de scraping.

Q2: NDJSON ou um array JSON — qual o formato que o scraper deve emitir?

NDJSON (um objeto por linha) carrega de forma mais limpa e faz streaming sem bufferizar o arquivo inteiro. Defina STRIP_OUTER_ARRAY = FALSE. Se seu produtor emitir um único array [ ... ], defina STRIP_OUTER_ARRAY = TRUE para que cada elemento se torne uma linha.

Q3: Devo carregar em uma coluna VARIANT ou colunas tipadas?

Armazene os dados brutos coletados em um VARIANT e molde-os com SQL — as páginas de origem mudam, e o VARIANT absorve novos campos sem necessidade de migração. Use MATCH_BY_COLUMN_NAME em colunas tipadas somente quando o esquema estiver estável.

Q4: Qual método de ingestão devo escolher?

COPY INTO para cargas únicas, Snowpipe para lotes de arquivos programados (minutos de latência, sem servidor), Snowpipe Streaming para frescor a nível de linha em menos de um minuto, e o conector Kafka quando os registros já fluem via Kafka. Comece com COPY INTO e depois avance conforme as necessidades de frescor aumentam.

Q5: Como lido com erros transitórios do scraper como os error 10054 ou um 503?

Trate-os como transitórios: feche a sessão Scrapeless, crie uma nova, navegue novamente e aguarde um seletor estável antes de extrair. Mantenha as tentativas limitadas. Esses erros pertencem ao lado do produtor e não afetam o Snowflake, que carrega o que chega ao estágio.

Q6: Preciso de um warehouse em funcionamento para o Snowpipe?

Não. O Snowpipe é sem servidor e cobrado por GB ingerido — o Snowflake fornece o processamento. Um warehouse gerenciado pelo usuário é necessário apenas para COPY INTO em massa e para consultas.

Q7: Posso rodar isso sem um agente de IA?

Sim. O CLI scrapeless-scraping-browser produz o NDJSON de ponta a ponta a partir de um shell simples, e o lado do Snowflake é SQL comum. Um agente conectado ao MCP é o caminho conveniente, não uma exigência.