Construa um Pipeline de Análise de Dados Raspados com Scrapeless e DuckDB
Advanced Data Extraction Specialist
TL;DR:
- Um pipeline de scraping que termina em um arquivo JSON é apenas metade de um pipeline; as perguntas analíticas surgem após os dados chegarem.
- DuckDB lê JSON Lines diretamente, então registros raspados se tornam uma tabela consultável sem servidor, sem migração de esquema e sem ferramenta ETL.
- Este pipeline busca 3 páginas através da API de Raspagem Universal Scrapeless, extrai 30 registros, carrega-os no DuckDB e escreve em Parquet.
- Parquet com compressão ZSTD armazenou os mesmos 30 registros em 4.674 bytes em comparação com 7.690 bytes de JSON Lines, e o DuckDB consulta o arquivo diretamente sem carregá-lo de volta.
- Cada etapa abaixo roda na sua máquina sem conta de warehouse na nuvem e sem credenciais além de uma chave Scrapeless.
- Comece no plano gratuito do Scrapeless e aponte a etapa de busca para sua própria fonte.
Pipeline em um Olhar
O fluxo é de cinco etapas, e a decisão de design interessante é onde os dados param de ser texto e começam a ser uma tabela.
fetch (Scrapeless) → descobrir registros → extrair campos → transformar em uma tabela tipada (DuckDB) → armazenar como Parquet
JSON Lines é o formato de transferência entre a metade de raspagem e a metade analítica. Ele é amigável para adição, sobrevive a uma execução travada sem corromper o que veio antes, e o DuckDB o lê nativamente — então não há carregador para escrever. O formato é especificado em a especificação do JSON Lines.
Nada aqui precisa de uma conta de warehouse. O DuckDB roda em processo, o que torna essa a forma mais barata de obter SQL real sobre dados raspados. Quando o destino é um warehouse gerenciado, o guia de ingestão do Snowflake cobre a mesma etapa de coleta com uma zona de aterrissagem diferente.
Pré-requisitos
- Python 3.9 ou posterior.
- Uma chave de API do Scrapeless do painel.
duckdbinstalado:
bash
pip install "duckdb==1.5.4"Defina a chave:
bash
export SCRAPELESS_API_KEY="sua_chave_api_aqui"Etapas 1–3: Buscar, Descobrir, Extrair
Uma chamada por página através da API de Raspagem Universal Scrapeless retorna HTML renderizado, e um parser da biblioteca padrão transforma cada bloco de citação em um registro. O extrator emite um objeto JSON por linha:
python
import json, os, urllib.request
from html.parser import HTMLParser
API = "https://api.scrapeless.com/api/v2/unlocker/request"
def fetch(url: str) -> str:
payload = json.dumps({
"actor": "unlocker.webunlocker",
"input": {"url": url, "js_render": True, "headless": True},
}).encode()
req = urllib.request.Request(
API, data=payload,
headers={"x-api-token": os.environ["SCRAPELESS_API_KEY"],
"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=120) as r:
return json.loads(r.read())["data"]
class QuoteParser(HTMLParser):
def __init__(self):
super().__init__()
self.rows, self._cur, self._cap = [], None, None
def handle_starttag(self, tag, attrs):
a = dict(attrs); cls = a.get("class", "")
if tag == "div" and "quote" in cls:
self._cur = {"text": "", "author": "", "tags": []}
elif self._cur is not None and tag == "span" and "text" in cls:
self._cap = "text"
elif self._cur is not None and tag == "small" and "author" in cls:
self._cap = "author"
elif self._cur is not None and tag == "a" and "tag" in cls:
self._cap = "tag"
def handle_data(self, data):
if self._cap == "text": self._cur["text"] += data
elif self._cap == "author": self._cur["author"] += data
elif self._cap == "tag": self._cur["tags"].append(data.strip())
def handle_endtag(self, tag):
if self._cap in ("text", "author", "tag"): self._cap = None
if tag == "div" and self._cur and self._cur["text"]:
self.rows.append(self._cur); self._cur = None
rows = []
for page in range(1, 4):
p = QuoteParser(); p.feed(fetch(f"https://quotes.toscrape.com/page/{page}/"))
rows.extend({"page": page, **r} for r in p.rows)
print(f"páginas buscadas: 3 | registros extraídos: {len(rows)}")
with open("quotes.jsonl", "w", encoding="utf-8") as f:
for r in rows: f.write(json.dumps(r, ensure_ascii=False) + "\n")
print(f"escreveu quotes.jsonl ({os.path.getsize('quotes.jsonl')} bytes)")
text
páginas buscadas: 3 | registros extraídos: 30
escreveu quotes.jsonl (7690 bytes)Duas opções aqui valem a pena manter na sua própria versão.
O número da página está anexado a cada registro no momento da extração. A proveniência é quase gratuita para registrar no ponto de coleta e cara para reconstruir posteriormente, e é o que permite responder "de qual página isso veio" sem ter que re-executar nada.
ensure_ascii=False mantém os caracteres tipográficos de citação intactos em vez de escapá-los. Isso importa quando o texto é o dado — a saída escapada ainda é JSON válido, mas inflaciona o arquivo e torna a inspeção posterior mais difícil.
Estágio 4: Transformar em uma Tabela Tipada
DuckDB lê o arquivo JSON Lines diretamente. read_json_auto infere o esquema, e o SELECT circundante é onde você impõe os tipos e colunas derivadas que realmente deseja:
python
import duckdb
con = duckdb.connect("quotes.duckdb")
con.execute("""
CREATE OR REPLACE TABLE quotes AS
SELECT
page::INTEGER AS page,
text AS quote_text,
author,
tags,
len(tags) AS tag_count
FROM read_json_auto('quotes.jsonl')
""")
total = con.sql("SELECT count(*) FROM quotes").fetchone()[0]
authors = con.sql("SELECT count(DISTINCT author) FROM quotes").fetchone()[0]
print(f"linhas carregadas: {total} | autores distintos: {authors}")
print(con.sql("""
SELECT author, count(*) AS quotes, round(avg(tag_count), 2) AS avg_tags
FROM quotes
GROUP BY author
ORDER BY quotes DESC, author
LIMIT 5
""").to_df().to_string(index=False))
con.execute("COPY quotes TO 'quotes.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")
import os
print(f"bytes parquet: {os.path.getsize('quotes.parquet')} | bytes jsonl: {os.path.getsize('quotes.jsonl')}")
rt = duckdb.sql("SELECT count(*) AS n, count(DISTINCT author) AS a FROM 'quotes.parquet'").fetchone()
print(f"ciclo completo de parquet -> linhas: {rt[0]}, autores: {rt[1]}")
text
linhas carregadas: 30 | autores distintos: 20
autor quotes avg_tags
Albert Einstein 6 2.83
J.K. Rowling 3 1.33
Bob Marley 2 1.00
Dr. Seuss 2 2.00
Marilyn Monroe 2 4.00
bytes parquet: 4674 | bytes jsonl: 7690
ciclo completo de parquet -> linhas: 30, autores: 20A coluna tags permanece uma lista em vez de ser achatada em uma string delimitada. DuckDB transporta tipos aninhados para o Parquet, de modo que len(tags) funciona em SQL e o array sobrevive ao ciclo completo — achatar para "a,b,c" neste estágio é um hábito prejudicial que vale a pena quebrar.
CREATE OR REPLACE TABLE torna a carga idempotente. Re-executar o estágio reconstrói a tabela a partir do arquivo atual em vez de anexar duplicatas, que é o comportamento desejado enquanto você ainda está iterando sobre o extrator.
O agregado é o ponto de todo o exercício: 30 registros, 20 autores distintos, e um autor responsável por 6 deles. Essa pergunta é uma consulta de uma linha contra uma tabela e um loop irritante contra um arquivo JSON.
Pronto para executar isso contra uma fonte que você se importa? Crie uma conta gratuita no Scrapeless e troque a URL no estágio de busca.
Estágio 5: Armazenar como Parquet
Os mesmos 30 registros ocupam 4.674 bytes como Parquet comprimido em ZSTD em comparação com 7.690 bytes como JSON Lines. Neste exemplo, a economia é modesta; o motivo para se importar é o que o formato faz em escala e o que permite depois.
Parquet é columnar e armazena os valores de cada coluna junto com sua própria codificação, que é definida em na especificação do formato de arquivo Apache Parquet. Uma consulta que toca duas das cinco colunas lê apenas aquelas duas. O codec de compressão usado aqui é especificado em no padrão de compressão Zstandard.
A linha do ciclo completo é o próprio teste do estágio. Ler o arquivo Parquet de volta retorna 30 linhas e 20 autores distintos, combinando com a tabela de onde veio — vale a pena afirmar em qualquer pipeline que escreve um arquivo que outro sistema lerá.
Note que a consulta final lê 'quotes.parquet' como uma tabela diretamente, sem passo de importação e sem conexão aberta com o arquivo de banco de dados. Essa é a razão prática para encerrar um pipeline de scraping em Parquet: a saída pode ser consultada pelo DuckDB e pela maioria dos outros mecanismos analíticos, exatamente onde está.
O Pipeline Completo
Executado como um único script, os cinco estágios são curtos o suficiente para serem lidos em uma única passada. Esta é a versão para copiar:
python
import json, os, urllib.request
from html.parser import HTMLParser
API = "https://api.scrapeless.com/api/v2/unlocker/request"
def fetch(url: str) -> str:
payload = json.dumps({
"actor": "unlocker.webunlocker",
"input": {"url": url, "js_render": True, "headless": True},
}).encode()
req = urllib.request.Request(
API, data=payload,
headers={"x-api-token": os.environ["SCRAPELESS_API_KEY"],
"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=120) as r:
return json.loads(r.read())["data"]
class QuoteParser(HTMLParser):
def __init__(self):
super().__init__()
self.rows, self._cur, self._cap = [], None, None
def handle_starttag(self, tag, attrs):
a = dict(attrs); cls = a.get("class", "")
if tag == "div" and "quote" in cls:
self._cur = {"text": "", "author": "", "tags": []}
elif self._cur is not None and tag == "span" and "text" in cls:
self._cap = "text"
elif self._cur is not None and tag == "small" and "author" in cls:
self._cap = "author"
elif self._cur is not None and tag == "a" and "tag" in cls:
self._cap = "tag"
def handle_data(self, data):
if self._cap == "text": self._cur["text"] += data
elif self._cap == "author": self._cur["author"] += data
elif self._cap == "tag": self._cur["tags"].append(data.strip())
def handle_endtag(self, tag):
if self._cap in ("text", "author", "tag"): self._cap = None
if tag == "div" and self._cur and self._cur["text"]:
self.rows.append(self._cur); self._cur = None
rows = []
for page in range(1, 4):
p = QuoteParser(); p.feed(fetch(f"https://quotes.toscrape.com/page/{page}/"))
rows.extend({"page": page, **r} for r in p.rows)
with open("quotes.jsonl", "w", encoding="utf-8") as f:
for r in rows: f.write(json.dumps(r, ensure_ascii=False) + "\n")
print(f"páginas buscadas: 3 | registros extraídos: {len(rows)}")
import duckdb
con = duckdb.connect("quotes.duckdb")
con.execute("""
CREATE OR REPLACE TABLE quotes AS
SELECT page::INTEGER AS page, text AS quote_text, author, tags, len(tags) AS tag_count
FROM read_json_auto('quotes.jsonl')
""")
rows_loaded = con.sql("SELECT count(*) FROM quotes").fetchone()[0]
authors = con.sql("SELECT count(DISTINCT author) FROM quotes").fetchone()[0]
print(f"linhas carregadas: {rows_loaded} | autores distintos: {authors}")
con.execute("COPY quotes TO 'quotes.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")
print(f"bytes parquet: {os.path.getsize('quotes.parquet')} | bytes jsonl: {os.path.getsize('quotes.jsonl')}")
rt = duckdb.sql("SELECT count(*), count(DISTINCT author) FROM 'quotes.parquet'").fetchone()
print(f"circuito completo de parquet -> linhas: {rt[0]}, autores: {rt[1]}")
## Onde este pipeline vai a seguir
**Particione por data de coleta** uma vez que você o execute repetidamente. Escrever em um layout `data/dt=<data-de-coleta>/quotes.parquet` permite que uma consulta ignore diretórios inteiros em vez de escanear a história.
**Mantenha os arquivos JSON Lines.** Eles são o registro bruto do que foi coletado; Parquet é o artefato derivado e tipado. Quando um bug do extrator surgir, você pode rederivar dos arquivos brutos em vez de re-coletar.
**Adicione assertivas entre os estágios.** Uma verificação de contagem entre a extração e o carregamento captura um seletor que parou de corresponder silenciosamente — o modo de falha que de outra forma aparece como uma queda silenciosa na contagem de linhas semanas depois.
Antes de apontar isso para uma fonte ao vivo, verifique seus termos e suas diretrizes `/robots.txt`, que seguem <a href="https://datatracker.ietf.org/doc/html/rfc9309" rel="nofollow"><strong>o padrão do Protocolo de Exclusão de Robôs</strong></a>. Mantenha a coleta em páginas públicas e em um intervalo de páginas limitado como o acima.
## Conclusão
A diferença entre um scraper e um pipeline de analytics é menor do que parece. JSON Lines como a passagem, DuckDB como o motor de consulta, e Parquet como o artefato armazenado cobrem isso com uma dependência e sem infraestrutura — 3 páginas para dentro, 30 linhas tipadas para fora, consultáveis no local.
O passo que a maioria dos pipelines pula é o último. Escrever Parquet e depois lê-lo de volta para confirmar que as contagens correspondem transforma um passo de armazenamento em um passo verificado, que é a diferença entre um arquivo que você tem e um arquivo em que você pode confiar.
[Comece com o plano gratuito do Scrapeless](https://app.scrapeless.com/passport/login?utm_source=website&utm_medium=blog&utm_campaign=universalscrapingapi&utm_term=scrapeless-duckdb-parquet-pipeline) para executar a etapa de busca contra seus próprios alvos, e revise a [precificação do Scrapeless](https://www.scrapeless.com/pt/pricing?utm_source=website&utm_medium=blog&utm_campaign=universalscrapingapi&utm_term=scrapeless-duckdb-parquet-pipeline) quando dimensionar um trabalho recorrente.
## FAQ
**Q: Por que DuckDB em vez de um armazém em nuvem para dados raspados?**
DuckDB é executado em processo, sem servidor, sem conta e sem necessidade de comunicação em rede, o que se encaixa na escala com que a maioria dos projetos de raspagem realmente opera. Ele lê JSON e Parquet nativamente, portanto, não há necessidade de escrever um carregador. Um armazém em nuvem ganha seu lugar quando várias equipes precisam de acesso concorrente às mesmas tabelas — e não quando um pipeline precisa de SQL sobre sua própria saída.
**P: Preciso achatar campos aninhados como listas de tags antes de carregar?**
Não, e achatar perde informações. O DuckDB suporta tipos de lista de ponta a ponta, então `tags` permanece como um array durante o carregamento, através de `len(tags)` no SQL, e durante a gravação e leitura de volta do Parquet. Colapsá-lo em uma string delimitada força cada consulta posterior a reanalisar.
**P: Por que escrever JSON Lines entre a raspagem e o carregamento?**
Isso desacopla as duas partes. A raspagem é a parte mais lenta e propensa a falhas; uma vez que os registros estão no disco um por linha, você pode reexecutar o carregamento e a transformação quantas vezes quiser, sem precisar rebuscar. Adicionar linha por linha também significa que uma execução que falha no meio deixa os registros anteriores intactos e legíveis.
**P: Quão menor é o Parquet em comparação com JSON Lines?**
Neste exemplo de 30 registros, 4.674 bytes contra 7.690 — cerca de 40% menor. Não leia demais nessa proporção neste tamanho, onde a sobrecarga do arquivo domina. A verdadeira vantagem do Parquet são as leituras colunares: uma consulta que toca duas das cinco colunas lê apenas essas, o que é importante em volumes onde o arquivo não cabe confortavelmente na memória.
**P: Posso consultar o arquivo Parquet sem carregá-lo em um banco de dados primeiro?**
Sim, e a última linha da fase de carregamento faz exatamente isso — `SELECT ... FROM 'quotes.parquet'` sem conexão de banco de dados aberta e sem importação. Isso é o que torna o Parquet um bom artefato final para um pipeline de raspagem: a saída permanece consultável onde está, pelo DuckDB e por outros motores analíticos.Na Scorretless, acessamos apenas dados disponíveis ao público, enquanto cumprem estritamente as leis, regulamentos e políticas de privacidade do site aplicáveis. O conteúdo deste blog é apenas para fins de demonstração e não envolve atividades ilegais ou infratoras. Não temos garantias e negamos toda a responsabilidade pelo uso de informações deste blog ou links de terceiros. Antes de se envolver em qualquer atividade de raspagem, consulte seu consultor jurídico e revise os termos de serviço do site de destino ou obtenha as permissões necessárias.
