Cómo transmitir datos web a Snowflake con Scrapeless y Snowpipe Streaming

Puntos Clave:

• Datos web raspados sin un esquema fijo en Snowflake. Scrapeless Scraping Browser renderiza una página en un navegador en la nube y emite JSON delimitado por nuevas líneas (NDJSON); Snowflake lo ingiere en una columna VARIANT, por lo que los nuevos campos nunca rompen la carga.
• Cuatro métodos de ingesta, una forma de datos. COPY INTO en bloque para cargas únicas, Snowpipe para lotes continuos basados en archivos, Snowpipe Streaming para filas de baja latencia y el conector Kafka para canalizaciones impulsadas por eventos — todos leen el mismo NDJSON que produce Scrapeless.
• La arquitectura de alto rendimiento de Snowpipe Streaming está generalmente disponible (GA desde septiembre de 2025). Escribe filas directamente a través de un SDK (Java, Python, Node.js) o REST, con canales, tokens de desplazamiento y recuperación exacta una vez — sin archivos en etapa.
• El esquema en lectura mantiene los datos raspados flexibles. Consulta una columna VARIANT con la notación col:field::type y desanida arrays con LATERAL FLATTEN — sin migración cuando la página fuente agrega un campo.
• La CLI de Snowflake (snow) es la herramienta actual. pip install snowflake-cli, y luego snow sql -f ingest.sql ejecuta toda la configuración desde un solo archivo.
• Gratis para comenzar. Las nuevas cuentas de Scrapeless incluyen tiempo de ejecución gratuito del Scraping Browser — regístrate en Sitio Web de Scrapeless.

Introducción: de la página renderizada a una tabla de Snowflake

Los equipos de análisis quieren cada vez más datos web — catálogos de productos, listados, reseñas, señales de mercado — en el mismo almacén que sus datos de primera parte, para que pueda unirse, modelar y alimentar BI. Snowflake es un destino común porque su tipo VARIANT almacena JSON semiestructurado de manera nativa y lo hace consultable con SQL.

La fricción es la brecha entre los dos sistemas. Las páginas raspadas son renderizadas por JavaScript y están detrás de defensas anti-bot; los datos a menudo llegan como JSON anidado cuya forma cambia a medida que el sitio fuente cambia. Los cargadores construidos a mano que asignan cada campo a una columna fallan la primera vez que la página agrega uno.

Esta publicación describe un flujo de trabajo centrado en terminal que cierra esa brecha. Scrapeless Scraping Browser maneja la parte de renderizado y de detección anti-bot y emite NDJSON; Snowflake lo ingiere de cuatro maneras diferentes dependiendo de qué tan fresco necesita estar el dato. El productor de ejemplo es el entorno de raspado público books.toscrape.com, así que cada comando a continuación es reproducible — el mismo patrón se aplica a objetivos más difíciles (ver las guías hermanas Mejores Raspadores de Zillow en 2026 y Mejores Raspadores de Amazon en 2026).

Lo Que Puedes Hacer Con Esto

• Construye un lakehouse de datos web. Raspas catálogos y listados en Snowflake y únelos a datos internos de ventas o inventario.
• Ejecuta instantáneas de mercado programadas. Coloca un nuevo archivo NDJSON por ejecución en un stage y deja que Snowpipe lo cargue automáticamente en minutos.
• Alimenta tableros de control en casi tiempo real. Transmite eventos raspados fila por fila con Snowpipe Streaming para frescura de sub-minuto.
• Conecta una infraestructura Kafka existente. Envía registros raspados a un tema y deja que el conector Kafka de Snowflake los coloque.
• Mantén el esquema flexible. Almacena el JSON sin procesar en VARIANT y dale forma en el momento de la consulta, por lo que los cambios en el sitio fuente nunca bloquean una carga.

En Scrapeless, solo accedemos a datos disponibles públicamente mientras cumplimos estrictamente con las leyes, regulaciones y políticas de privacidad de los sitios web aplicables. El contenido de esta publicación es solo para fines de demostración.

Por Qué Scrapeless Scraping Browser

Scrapeless Scraping Browser es un navegador en la nube personalizable y anti-detección diseñado para rastreadores web y agentes de IA. Como el lado productor de un pipeline de Snowflake, ofrece:

• Renderizado de JavaScript del lado de la nube, de modo que los datos estén presentes en el DOM antes de la extracción
• Proxies residenciales en más de 195 países, fijados por sesión
• Huellas digitales de navegador anti-detección
• Una única superficie CLI scrapeless-scraping-browser cuya eval devuelve JSON que puedes reconfigurar en NDJSON en un solo paso
• Persistencia de sesión para raspados de múltiples páginas

Obtén tu clave API en el plan gratuito en Sitio Web de Scrapeless.

Requisitos Previos

• Node.js 18 o más reciente
• Una cuenta de Scrapeless y clave API — regístrate en Sitio Web de Scrapeless
• Una cuenta de Snowflake con un rol que pueda crear bases de datos, almacenes, etapas y conductos
• La CLI de Snowflake: pip install snowflake-cli (Python 3.10+)
• Para Snowpipe auto-ingest y etapas externas: un bucket en la nube (AWS S3, GCS o Azure) y permiso para crear una integración de almacenamiento
• jq es opcional (se muestra un fallback de una sola línea en Node para la conversión NDJSON)

Configura una conexión de Snowflake una vez, en ~/.snowflake/config.toml:

[connections.demo]
account   = "miempresa-mi_cuenta"
user      = "jondoe"

password = "tu_contraseña_aquí"
warehouse = "ingest_wh"
database = "web_data"
schema = "raw"
role = "sysadmin"

Luego, `snow sql -c demo -q "SELECT CURRENT_VERSION();"` confirma que funciona.

---

## El pipeline a grandes rasgos

Scraping sin esfuerzo Browser → archivo NDJSON → etapa Snowflake → tabla (VARIANT)
(renderizar + extraer) (un objeto (interno o COPY INTO | cargado único
por línea) externo) Snowpipe | continuo
Streaming | baja latencia
Kafka | basado en eventos

La forma nunca cambia: Scrapeless emite un objeto JSON por línea, el archivo se coloca en una etapa y uno de los cuatro métodos lo carga en una columna `VARIANT` que consultas con SQL.

---

## Paso 1 — Producir NDJSON con Scrapeless

Instala el CLI y configura tu clave:

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

Abre una sesión en la nube, navega a la página del catálogo, espera un marcador estable y extrae los registros de libros con eval. El JSON new-session anida el id bajo data.taskId — usa jq, o la alternativa portable de grep mostrada:

# abre una sesión y captura el id de tarea (la ruta jq es .data.taskId)
SID=$(scrapeless-scraping-browser new-session --name books --ttl 300 --proxy-country US --json | jq -r '.data.taskId')
# ¿sin jq? alternativa portable:
# SID=$(scrapeless-scraping-browser new-session --name books --ttl 300 --proxy-country US --json | grep -oE '"taskId":"[^"]*"' | head -1 | cut -d'"' -f4)

# renderiza la página del catálogo, luego espera la cuadrícula de productos
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"

# extrae un registro por libro; el eval devuelve un 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

Convierte el array a NDJSON — un objeto por línea, que es el formato que los cargadores de Snowflake leen más claramente:

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

# o, sin jq, una línea de código de Node
node -e 'JSON.parse(require("fs").readFileSync("books.raw.json","utf8")).forEach(o=>console.log(JSON.stringify(o)))' > books.ndjson

books.ndjson ahora contiene un objeto JSON auto-contenido por línea. Si una sesión fría devuelve un shell vacío o un transitorio os error 10054, cierra la sesión, crea una nueva y vuelve a intentar un número limitado de veces antes de extraer.

Paso 2 — Preparar Snowflake

Crea el almacén, la base de datos, el esquema, un formato de archivo JSON y una tabla de aterrizaje con una sola columna VARIANT. Guarda esto como setup.sql y ejecútalo con 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: un objeto JSON por línea, así que no quites un array externo
CREATE OR REPLACE FILE FORMAT ndjson_format
  TYPE = JSON
  STRIP_OUTER_ARRAY = FALSE
  COMPRESSION = AUTO;

-- aterriza el registro bruto tal como está; dale forma en el tiempo de consulta
CREATE OR REPLACE TABLE raw_books (
  src       VARIANT,
  loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

STRIP_OUTER_ARRAY = FALSE es correcto para NDJSON porque cada línea ya es su propio objeto — STRIP_OUTER_ARRAY = TRUE es solo para un archivo que es un gran [ ... ] array.

Paso 3 — Método 1: Carga masiva única con COPY INTO

Para un solo archivo o un lote manual, coloca el archivo en la etapa y ejecuta COPY INTO. El camino más simple es una etapa interna nombrada más PUT:

-- una etapa interna nombrada vinculada al formato JSON
CREATE OR REPLACE STAGE books_stage FILE_FORMAT = ndjson_format;

# sube el NDJSON local a la etapa interna (el CLI de snow ejecuta PUT)
snow sql -c demo -q "PUT file://$(pwd)/books.ndjson @books_stage AUTO_COMPRESS=TRUE OVERWRITE=TRUE"

-- carga cada objeto como una fila en la columna VARIANT
COPY INTO raw_books (src)
  FROM @books_stage
  FILE_FORMAT = (FORMAT_NAME = 'ndjson_format')
  ON_ERROR = 'CONTINUE';

Para mapear claves JSON directamente a columnas tipadas en lugar de un VARIANT, crea una tabla cuyos nombres de columna coincidan con las claves y utiliza MATCH_BY_COLUMN_NAME:

CREATE OR REPLACE TABLE books (
  title VARCHAR, price VARCHAR, rating VARCHAR, in_stock BOOLEAN, url VARCHAR
);

COPY INTO books
  FROM @books_stage
  FILE_FORMAT = (TYPE = 'JSON')
  MATCH_BY_COLUMN_NAME = 'CASE_INSENSITIVE';

Si prefieres permitir que Snowflake derive el esquema de los archivos en etapa, INFER_SCHEMA con CREATE TABLE … USING TEMPLATE construye la lista de columnas para ti:

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

Para los datos que ya están en un bucket de la nube, apunta a un stage externo en lugar de cargarlo. Con una integración de almacenamiento (sin claves en línea):

CREATE OR REPLACE STAGE books_s3_stage
  URL = 's3://my-bucket/scraped/books/'
  STORAGE_INTEGRATION = my_s3_integration
  FILE_FORMAT = ndjson_format;

COPY INTO raw_books (src) FROM @books_s3_stage;

Obtén tu clave API en el plan gratuito: app.scrapeless.com

Paso 4 — Método 2: Lotes continuos con Snowpipe

Cuando el scraper coloca un nuevo archivo en un bucket según un horario, Snowpipe carga automáticamente cada archivo — sin COPY manual y sin un warehouse dedicado. Una tubería envuelve una declaración COPY INTO; con AUTO_INGEST = TRUE, una notificación de evento en la nube activa la carga:

CREATE OR REPLACE PIPE books_pipe
  AUTO_INGEST = TRUE
  AWS_SNS_TOPIC = 'arn:aws:sns:us-east-1:123456789012:scraped-bucket'
  AS
  COPY INTO raw_books (src)
  FROM @books_s3_stage
  FILE_FORMAT = (TYPE = 'JSON');

En S3, el evento fluye a través de SNS/SQS en una cola administrada por Snowflake; GCS utiliza Pub/Sub y Azure utiliza Event Grid, cada uno conectado con una integración de notificación. Si prefieres llamar a Snowpipe explícitamente, deja AUTO_INGEST sin configurar y POSTea las rutas de archivos en etapa al endpoint REST insertFiles, luego consulta insertReport.

Dos notas operativas de la guía de Snowflake:

• La facturación es sin servidor y ahora se cobra por cada GB de datos que Snowpipe ingiere — no hay que dimensionar un warehouse, y el antiguo componente por archivo ha sido retirado.
• El tamaño de los archivos importa. Apunta a archivos de alrededor de 100–250 MB comprimidos, y no realices más de una carga por minuto; realizar cargas más frecuentes añade sobrecarga de gestión de cola sin reducir la latencia. Agrupa pequeños lotes de raspado en archivos más grandes antes de cargar.

Snowpipe hace que los datos estén disponibles en minutos, lo que se ajusta bien a instantáneas de mercado programadas.

Paso 5 — Método 3: Filas de baja latencia con Snowpipe Streaming

Cuando la frescura necesita ser de segundos, no de minutos, Snowpipe Streaming escribe filas directamente en una tabla — sin archivos en etapa. La arquitectura de alto rendimiento ha estado generalmente disponible desde septiembre de 2025, con SDK para Java, Python y Node.js, además de una API REST sobre un núcleo de cliente compartido; el Snowpipe Streaming basado en archivos clásico está en un camino de depreciación.

El modelo tiene tres conceptos fundamentales:

• Canales — una conexión de transmisión nombrada y de larga duración a una tabla. Las filas se comprometen en orden dentro de un canal.
• Tokens de desplazamiento — una cadena que tu aplicación adjunta a cada lote. Después de un reinicio, getLatestCommittedOffsetToken() te dice la última posición comprometida de manera duradera, para que solo reproduzcas lo que sigue — la base para la entrega exactamente una vez.
• Facturación por rendimiento — créditos por GB sin comprimir ingeridos, en lugar de por archivo.

La forma del cliente de Java es pequeña:

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

OpenChannelRequest request = OpenChannelRequest.builder("BOOKS_CHANNEL")
    .setDBName("WEB_DATA").setSchemaName("RAW").setTableName("RAW_BOOKS")
    .setOnErrorOption(OpenChannelRequest.OnErrorOption.CONTINUE)
    .build();

SnowflakeStreamingIngestChannel channel = client.openChannel(request);
channel.insertRow(rowAsMap, offsetToken);   // un registro raspado
channel.getLatestCommittedOffsetToken();     // para recuperación

Utiliza Streaming cuando los registros raspados llegan como un flujo continuo (un agente emitiendo eventos mientras rastrea) y el panel necesita esos datos en segundos.

Paso 6 — Método 4: Cargas impulsadas por eventos con el conector Kafka

Si los registros raspados ya fluyen a través de Apache Kafka, el Conector de Snowflake para Kafka lleva un tema a una tabla (un tema se asigna a una tabla). Se ejecuta dentro de un trabajador de Kafka Connect. Una sola propiedad selecciona el motor de ingestión subyacente:

name=scraped-books-sink
connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
topics=scraped_books
snowflake.database.name=WEB_DATA
snowflake.schema.name=RAW
# SNOWPIPE (basado en archivos, por defecto) o SNOWPIPE_STREAMING (baja latencia)
snowflake.ingestion.method=SNOWPIPE_STREAMING

Cada tabla que el conector crea tiene dos columnas VARIANT: RECORD_CONTENT (la carga del mensaje) y RECORD_METADATA (tema, partición, desplazamiento, marcas de tiempo y encabezados). Consulta la carga útil exactamente como lo harías con cualquier VARIANT.

Para una alternativa completamente gestionada — sin un clúster de Kafka Connect para operar — Snowflake Openflow (disponible de forma general, construido sobre Apache NiFi) ingiere de Kafka, Kinesis, bases de datos y fuentes SaaS en Snowflake a través de tuberías gestionadas.

Elección de un método

La mayoría de los equipos comienzan con COPY INTO para validar el esquema, pasan a Snowpipe una vez que el raspador funciona en un horario, y adoptan Streaming o Kafka solo cuando la frescura de menos de un minuto justifica la pieza adicional en movimiento.

Consultando los datos cargados

Debido a que el registro en bruto vive en un VARIANT, lo moldeas en el momento de lectura. Navega con el operador : y convierte con :::

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;

Cuando un registro raspado lleva un array — una lista de photos, una serie de priceHistory — LATERAL FLATTEN lo descompone en filas:

-- raw_listings: una tabla hipotética de listados raspados cargada de la misma manera que raw_books.
-- (El ejemplo de libros no tiene un array anidado; esto muestra el patrón para una fuente que sí tiene.)
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;

No se necesita migración cuando la página fuente añade un campo — simplemente aparece bajo src en la siguiente carga.

Lo que recibes de vuelta

Después de COPY INTO, cada objeto raspado es una fila en raw_books. La columna VARIANT contiene el registro tal cual; el esquema a continuación es normativo y los valores de los campos son muestras ilustrativas:

// El VARIANT src de una fila, tal como lo emitió el extractor del Paso 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"
}

Algunas observaciones honestas:

• Los precios llegan como cadenas de visualización. Convierte y limpia en SQL (REPLACE(src:price::string, '£', '')::number) en lugar de esperar números de la página.
• Los campos condicionales son anulables. Un campo ausente en una página dada simplemente falta en src; el acceso a VARIANT devuelve NULL en lugar de una error.
• MATCH_BY_COLUMN_NAME omite claves no coincidentes. Nuevas claves llegan automáticamente a una tabla VARIANT pero son descartadas por una tabla tipada hasta que añadas la columna.
• El historial de carga de Snowpipe se retiene 14 días en los metadatos del pipe; el historial de COPY a granel se retiene 64 días en los metadatos de la tabla — ten eso en cuenta al auditar un relleno.

Conclusión

Obtener datos web raspados en Snowflake se reduce a cuatro movimientos: renderizar y extraer con Scrapeless, emitir NDJSON, colocar el archivo y cargarlo con el método cuya latencia se ajuste — COPY INTO, Snowpipe, Snowpipe Streaming, o el conector de Kafka. Coloca el registro en bruto en una columna VARIANT y moldea en el momento de la consulta, para que la tubería sobreviva a la adición de campos en el sitio fuente.

Pin US egress cuando el objetivo lo necesite, mantén la cadena de sesión Scrapeless dentro de una invocación de shell, sigue el patrón de descubrimiento → extracción, y trata los campos ausentes como anulables. Para objetivos más difíciles que el sandbox utilizado aquí, el mismo patrón de productor se traslada — consulta la guía hermana Mejores raspadores de Zillow en 2026, la página del producto de Scraping Browser, y la documentación de Scrapeless.

¿Listo para construir tu canal de datos potenciado por IA?

Únete a nuestra comunidad para reclamar un plan gratuito y conectar con desarrolladores que construyen tuberías de web-datos-a-almacenamiento: Discord · Telegram.

Regístrate en Sitio web de Scrapeless para obtener un tiempo de ejecución gratuito de Scraping Browser, y consulta scrapeless.com/en/pricing para escalar minutos de sesión y concurrencia a medida que crece la tubería.

FAQ

Q1: ¿Es legal el raspado web para la ingestión en almacenes?
Recolección de datos visibles públicamente es ampliamente defensable, pero la legalidad depende de los términos del sitio objetivo, la jurisdicción y el tipo de datos. Revisa los términos de servicio del sitio, evita datos personales o restringidos, y consulta a un abogado antes de su uso comercial. El sitio de prueba utilizado aquí, books.toscrape.com, existe específicamente para la práctica de scraping.

Q2: ¿NDJSON o un array JSON — cuál debería emitir el scraper?

NDJSON (un objeto por línea) se carga de la manera más limpia y se transmite sin almacenar todo el archivo en búfer. Establece STRIP_OUTER_ARRAY = FALSE. Si tu productor emite un único array [ ... ], establece STRIP_OUTER_ARRAY = TRUE para que cada elemento se convierta en una fila.

Q3: ¿Debería cargar en una columna VARIANT o en columnas tipadas?

Almacena datos crudos raspados en un VARIANT y dales forma con SQL — las páginas de origen cambian, y VARIANT absorbe nuevos campos sin migración. Usa MATCH_BY_COLUMN_NAME en columnas tipadas solo una vez que el esquema esté estable.

Q4: ¿Qué método de ingestión debería elegir?

COPY INTO para cargas únicas, Snowpipe para lotes de archivos programados (minutos de latencia, sin servidor), Snowpipe Streaming para frescura a nivel de fila a sub-minuto, y el conector de Kafka cuando los registros ya fluyen sobre Kafka. Comienza con COPY INTO, luego gradúate a medida que crezcan las necesidades de frescura.

Q5: ¿Cómo manejo errores transitorios del scraper como os error 10054 o un 503?

Trátalos como transitorios: cierra la sesión de Scrapeless, crea una nueva, navega de nuevo y espera un selector estable antes de extraer. Mantén los reintentos acotados. Estos pertenecen al lado del productor y no afectan a Snowflake, que carga lo que llega a la etapa.

Q6: ¿Necesito un almacén en funcionamiento para Snowpipe?

No. Snowpipe es sin servidor y se factura por GB ingerido — Snowflake proporciona el procesamiento. Un almacén gestionado por el usuario solo es necesario para COPY INTO en bloque y para consultas.

Q7: ¿Puedo ejecutar esto sin un agente de IA?

Sí. La CLI scrapeless-scraping-browser produce el NDJSON de principio a fin desde una terminal común, y el lado de Snowflake es SQL ordinario. Un agente conectado a MCP es el camino conveniente, no un requisito.