De la API al warehouse: construyendo un pipeline de datos en R

Quería monitorear un marketplace de forma sistemática. No mirar precios a mano ni copiar tendencias en una planilla, sino construir un pipeline que todos los días extrajera datos, los normalizara y los dejara listos para análisis. La idea era simple: keywords de tendencia, productos asociados, precios, atributos. Un snapshot diario que fuera acumulando historia.

La pregunta era con qué construirlo.

Por qué R y no Python

La respuesta honesta: porque ya tenía la infraestructura mental. Años trabajando con tidyverse, DBI, RPostgres. Podría haberlo hecho en Python con requests y pandas, pero R tiene algo que a menudo se subestima para ETL: httr2 es extraordinariamente expresivo para APIs REST, y R6 permite diseñar módulos con estado interno de forma limpia.

El pipeline quedó como un paquete R con cinco clases R6:

MercadoLibre/
├── R/
│   ├── MLAuth.R              # OAuth 2.0, refresh automático
│   ├── TrendExtractor.R      # Keywords de tendencia por categoría
│   ├── CatalogExtractor.R    # Detalle de productos del catálogo
│   ├── GoogleTrendsExtractor.R  # Validación con volumen de búsqueda
│   └── NeonConnector.R       # Persistencia idempotente a PostgreSQL
├── scripts/
│   └── run_pipeline.R        # Orquestador: ejecuta los 7 pasos
├── Dockerfile
├── entrypoint.sh
└── setup.sh

Cada clase encapsula un dominio: autenticación, extracción de tendencias, catálogo de productos, validación externa, y persistencia. El orquestador simplemente las instancia y las llama en orden.

OAuth 2.0 con rotación automática de tokens

La API del marketplace usa OAuth 2.0 con tokens que expiran. El access token dura 6 horas; el refresh token se invalida cada vez que se usa. Eso significa que el pipeline tiene que rotar el refresh token después de cada ejecución, o al día siguiente no puede autenticarse.

La clase MLAuth maneja todo el flujo:

auth <- MLAuth$new(
  client_id     = Sys.getenv("CLIENT_ID"),
  client_secret = Sys.getenv("CLIENT_SECRET"),
  redirect_uri  = Sys.getenv("REDIRECT_URI")
)
auth$refresh()  # Renueva access_token usando refresh_token

Después de ejecutar el pipeline, el entrypoint.sh lee el nuevo refresh token del .env y lo actualiza en Secret Manager via REST API. Si este paso falla, el pipeline del día siguiente muere. Aprendí esto de la peor forma posible.

Idempotencia por snapshot_date

El patrón más importante del pipeline es la idempotencia. Si el Job falla a mitad de camino y lo re-ejecuto, no quiero datos duplicados ni inconsistencias. La solución fue usar snapshot_date como clave de transacción:

write_trends = function(df, snapshot_date) {
  DBI::dbWithTransaction(self$con, {
    DBI::dbExecute(self$con,
      "DELETE FROM trends WHERE snapshot_date = $1",
      params = list(snapshot_date)
    )
    DBI::dbWriteTable(self$con, "trends", df, append = TRUE, row.names = FALSE)
  })
}

DELETE + INSERT dentro de una transacción. Si ejecuto el pipeline dos veces el mismo día, el resultado en la base de datos es idéntico. Cuatro tablas — trends, products, product_attributes, google_trends — todas siguen este patrón.

Cloud Run Jobs: el cron serverless

El pipeline corre como un Cloud Run Job, no como un Cloud Run Service. La diferencia es crucial: un Job ejecuta, termina y se apaga. No tiene endpoint HTTP, no escucha requests, no cobra por tiempo idle. Perfecto para ETL.

La configuración del scheduler es un cron que dispara el Job todos los días:

gcloud scheduler jobs create http my-pipeline-trigger \
  --schedule="0 7 * * *" \
  --time-zone="America/Santiago" \
  --uri="https://REGION-run.googleapis.com/..." \
  --http-method=POST

El Dockerfile está basado en rocker/r-ver:4.4 con los paquetes del sistema necesarios para PostgreSQL y SSL. El setup.sh automatiza todo: crea los secrets, build de la imagen, crea el Job y el Scheduler.

La pesadilla del CRLF

Windows usa \r\n como fin de línea. Linux usa \n. El entrypoint.sh del contenedor tiene un shebang (#!/bin/bash) que, con un \r invisible al final, se convierte en #!/bin/bash\r — y bash no encuentra ese intérprete.

El error era críptico:

/usr/bin/env: 'bash\r': No such file or directory

La solución fue un .gitattributes que fuerza line endings Unix para todos los scripts:

analytics/mercadolibre/*.sh eol=lf
analytics/mercadolibre/R/*.R eol=lf

Un archivo de dos líneas que me ahorró horas de debugging futuro. Si trabajas en Windows y despliegas en Linux, .gitattributes no es opcional.

Los primeros datos

Después de resolver el OAuth, la idempotencia, el CRLF, los timeouts (el Job necesitó 4 horas de timeout, no la hora default) y el rate limiting (exponential backoff con respeto al header Retry-After), el pipeline corrió por primera vez.

Las tablas se llenaron. Tendencias con ranking, productos con precios y atributos, todo con su snapshot_date. Datos reales, actualizándose solos cada mañana.

Pero al mirar los datos me di cuenta de que el pipeline extraía demasiado ruido y poco signal. Eso lo cuento en la próxima entrega, donde rediseñé la extracción y construí el dashboard para visualizarlo todo.