API de sincronizaciones de warehouse

Importaciones programadas desde tu data warehouse - BigQuery, Snowflake o Redshift - hacia las mismas rutas de escritura que usa una importación CSV.

Todos los endpoints de esta página son relativos a la URL base: https://api-eu1.joryio.com/api - consulta Visión general de la API.

Visión general

Una sincronización lee una tabla o vista que tú preparas y escribe cada página de filas en Joryio a medida que llega. Nada se acumula en memoria, así que sincronizar diez millones de filas cuesta la misma memoria que sincronizar diez mil.

Dos cosas que una sincronización no es:

• No es SQL libre. La consulta se genera en cada ejecución para poder incorporar el predicado incremental, así que nos das el nombre de una tabla o vista y una columna de marca de agua, no un SELECT.
• No es una vía de ingesta aparte. Las filas entran exactamente por el mismo código que usa una importación de archivo, incluidas las reglas de transición de consentimiento: una baja en tu warehouse se respeta, y una re-suscripción no, salvo que lo indiques deliberadamente.

Autenticación

Authorization: Bearer YOUR_API_KEY

Leer una sincronización requiere integrations:read; crear, actualizar, ejecutar o eliminar una requiere integrations:write. Activar overrideConsent requiere además compliance:write.

El objeto de sincronización

Campo	Tipo	Notas
integrationId	uuid	La conexión de warehouse guardada por la que leer
connectorId	string	bigquery, snowflake o redshift
syncType	string	users, events o deletions - decide qué ruta de escritura siguen las filas
sourceTable	string	table, schema.table o db.schema.table. Este campo o sourceQuery
sourceQuery	string?	Un SELECT de solo lectura guardado, en lugar de una tabla
watermarkColumn	string?	La columna que indica que una fila cambió. Omítela para un refresco completo en cada ejecución
keyColumn	string?	Un desempate único y ordenable. Recomendado - ver más abajo
watermark	timestamp?	Hasta dónde ha leído la sincronización. Lo gestionamos nosotros; bórralo con Rewind
columnMapping	object	Columna del warehouse a ruta del campo destino
schedule	string	manual, every_5_min, every_15_min, hourly, every_6_hours, daily, weekly, monthly
cadence	object	Solo lectura. configuredMs, effectiveMs, throttled - ver más abajo
enabled	boolean	Desactivada significa que el planificador la omite, conservando la programación para después
overrideConsent	boolean	Trata el warehouse como autoritativo para el consentimiento. Requiere compliance:write
lastRunStats	object?	Filas leídas, creadas, actualizadas, omitidas, fallidas, duración y una muestra acotada de errores

Mapeo de columnas

Asigna cada columna del warehouse a una ruta de campo destino:

Destino	Significado
email, phone, externalId	Identidad. Se requiere al menos email o externalId por fila
attributes.<key>	Un atributo de perfil. La clave no puede contener un punto
subscriptions.<channel>	Consentimiento global de canal - email, sms, whatsapp, push, viber
subscriptions.list.<listId>.<channel>	Consentimiento en UN grupo de suscripción
userId, eventName, timestamp, properties.<key>	Para syncType: events

Eliminaciones

Una sincronización deletions lee una tabla de personas a borrar. No elimina nada por sí misma: crea solicitudes de supresión por la misma vía de cumplimiento que usa la API de DSR, de modo que una supresión recorre perfiles, suscripciones por lista, eventos, pases de wallet y archivos almacenados, respeta tu ajuste de anonimizar-en-lugar-de-borrar y deja el rastro de auditoría que se le mostraría a un regulador.

Dos consecuencias que conviene conocer:

• Una fila que nombra a alguien que nunca se importó, o que ya fue borrado, se omite, no falla. Las listas de eliminación contienen ambos casos de forma rutinaria.
• Alguien que ya está siendo borrado no se vuelve a encolar. La fila permanece en tu tabla hasta que la quites, así que sin esa protección cada ejecución encolaría otra supresión para la misma persona.

Consentimiento

El consentimiento se mueve a través de una sincronización con las mismas reglas que una importación de archivo:

• Una baja siempre se respeta, sin permiso ni indicador. Sincronizar exclusiones desde tu CRM no necesita activar nada.
• Una sincronización no volverá a suscribir a quien se dio de baja, y no degradará un doble opt-in confirmado, salvo que overrideConsent esté activo.

overrideConsent significa más en una sincronización que en un archivo. Un archivo se aplica una vez; una sincronización reaplica su origen en cada ciclo, así que una anulación dejada activa revierte cada baja que llegue entre ejecuciones, mientras la sincronización exista.

Endpoints

Listar sincronizaciones

GET /warehouse-syncs

Obtener una sincronización

GET /warehouse-syncs/:id

Devuelve la sincronización, incluidos lastRunStats y el watermark actual.

Previsualizar filas de origen

POST /warehouse-syncs/preview

{ "integrationId": "0f8e...", "sourceTable": "analytics.public.contacts", "limit": 20 }

Una lectura acotada del origen tal como está - sin marca de agua, sin predicado incremental - para que construyas el mapeo contra los nombres reales de las columnas y no contra los recordados. Envía exactamente uno de sourceTable o sourceQuery. Se ejecuta en línea y está limitado a 100 filas como máximo.

Escribir SQL en lugar de nombrar una tabla

sourceQuery acepta un SELECT de solo lectura guardado, para orígenes que una vista no puede dar forma. Se comprueba con la misma protección de solo lectura cuando lo guardas, no a las 3 de la mañana la primera vez que se ejecuta.

El predicado incremental se aplica envolviendo tu consulta:

SELECT * FROM (<tu consulta>) AS jry_src WHERE updated_at > :watermark ORDER BY updated_at ASC

Por tanto tu consulta debe devolver la columna de marca de agua, y la columna clave si defines una. No las alias tú: nombra la columna en watermarkColumn y asegúrate de que tu SELECT la devuelve. Si no lo hace, guardar la sincronización falla con un mensaje que enumera las columnas que tu origen devuelve realmente, en vez de fallar en su primera ejecución a las 3 de la mañana. Ese contrato es la razón de que una tabla o vista sea lo predeterminado y esto sea la salida de emergencia. Una sola sentencia: una segunda se ejecutaría fuera de la subconsulta.

Listar ejecuciones

GET /warehouse-syncs/:id/runs?limit=50

Ejecuciones recientes, de más nueva a más antigua. Cada una lleva sus contadores, la marca de agua con la que empezó y terminó, una muestra acotada de errores por fila y un error de ejecución si falló la ejecución en sí.

El objeto de sincronización te dice si una sincronización funciona ahora. Esto te dice cuándo empezó a fallar, que es la pregunta que realmente tienes cuando los números parecen raros: una sincronización que lleva un mes omitiendo el 3% de sus filas se ve idéntica, en el objeto, a una que omitió el 3% esta mañana.

Se conservan las 50 ejecuciones más recientes por sincronización. La muestra de errores está acotada porque un mapeo de columnas roto hace fallar todas las filas, y un millón de mensajes idénticos no ayuda a nadie: los contadores son la verdad sobre la escala, la muestra es la verdad sobre la causa.

Crear una sincronización

POST /warehouse-syncs

{
  "integrationId": "0f8e...",
  "connectorId": "snowflake",
  "name": "Contacts sync",
  "syncType": "users",
  "sourceTable": "analytics.public.contacts",
  "watermarkColumn": "updated_at",
  "schedule": "hourly",
  "columnMapping": {
    "EMAIL": "email",
    "CUSTOMER_ID": "externalId",
    "PLAN": "attributes.plan",
    "EMAIL_CONSENT": "subscriptions.email"
  }
}

sourceTable y watermarkColumn deben ser identificadores simples. Cualquier otra cosa - comillas, espacios, puntos y coma, marcadores de comentario - se rechaza con un 400, porque esos nombres se interpolan en el SQL generado.

Actualizar una sincronización

PATCH /warehouse-syncs/:id

Crear y actualizar solo aceptan los campos que puedes definir: name, integrationId, syncType, sourceTable, sourceQuery, watermarkColumn, keyColumn, columnMapping, schedule, enabled, overrideConsent. Cualquier otro campo - id, organizationId, workspaceId, status, watermark o los contadores de ejecución - se rechaza con un 400; esos los gestiona Joryio. integrationId debe ser una conexión de almacén de datos del mismo espacio de trabajo que la sincronización; de lo contrario, la solicitud devuelve 404. connectorId es opcional y se toma de esa conexión - si lo envías, debe coincidir.

Cambiar schedule o enabled recalcula cuándo toca la siguiente ejecución.

Ejecutar ahora

POST /warehouse-syncs/:id/run

Devuelve { "success": true, "queued": true }, o queued: false cuando la sincronización ya está encolada o en ejecución. Una sincronización que se solapa consigo misma relee las mismas filas y duplica la factura del warehouse sin datos nuevos.

Rewind

POST /warehouse-syncs/:id/rewind

Borra la marca de agua para que la siguiente ejecución relea la tabla desde el principio. Es seguro - cada escritura es un upsert idempotente con clave de identidad - y es el remedio cuando un mapeo de columnas era incorrecto: corrige el mapeo, rebobina, ejecuta.

Eliminar

DELETE /warehouse-syncs/:id

Elimina la configuración. Los datos ya importados no se tocan.

Cadencia efectiva

Una ejecución que tarda más que su intervalo se rechaza, no se encola: una sincronización nunca se solapa consigo misma. Así que una programación de 5 minutos sobre una sincronización que tarda 7 se ejecuta realmente cada 7 minutos, y cadence.throttled es como te enteras en lugar de preguntártelo.

every_5_min es el suelo práctico para una lectura por sondeo, y la razón es el coste más que la capacidad: Snowflake factura por segundo de warehouse activo con un mínimo de 60 segundos en cada reanudación, así que una sincronización tan frecuente nunca deja dormir al warehouse. BigQuery factura bytes escaneados, y el predicado incremental solo evita un escaneo completo cuando la tabla está particionada o agrupada por la columna de marca de agua.

Mensual

monthly conserva el día del mes en que empezó la sincronización. Una sincronización anclada al 31 se ejecuta el 28 de febrero (29 en año bisiesto) y vuelve al 31 de marzo: no migra al 28 y se queda ahí. Toda la aritmética de programación es UTC, porque una sincronización no lleva zona horaria y la aritmética en hora local significaría silenciosamente la zona en la que resulte estar corriendo el planificador.

Cómo funcionan las lecturas incrementales

Con una watermarkColumn definida, cada ejecución lee las filas iguales o posteriores a la marca de agua almacenada, ordenadas por esa columna. La marca de agua solo avanza después de que una página se haya escrito de forma duradera, así que una ejecución interrumpida relee su cola en lugar de saltársela.

Con una keyColumn, el problema del límite desaparece en vez de gestionarse. El cursor pasa a ser (watermarkColumn, keyColumn) y cada ejecución lee estrictamente después de la última fila leída:

WHERE updated_at > :ts OR (updated_at = :ts AND id > :lastId)
ORDER BY updated_at, id

Las filas que comparten marca de tiempo se leen exactamente una vez - ni se pierden ni se releen - y no hace falta una consulta de conteo adicional. Elige algo monótono: un id autoincremental o un ULID. Un hash de contenido es estable pero de orden aleatorio, así que una fila insertada después en el mismo instante se ordenaría antes del cursor y no se leería nunca.

Sin keyColumn recurrimos a un cursor solo de marca de tiempo, y el límite se trata de una de dos formas:

• Huellas (por defecto, hasta 5.000 filas en el límite). Recordamos qué filas estaban en el instante límite y omitimos exactamente esas en la siguiente ejecución. El límite se relee cada vez para poder reconocerlas, y no se pierde nada.
• Conteos de filas (por encima de 5.000). Recordamos cuántas filas había y solo releemos cuando ese número crece. Más barato, porque el límite no se relee, y tiene un agujero: una fila borrada y una insertada en el mismo instante dejan el conteo igual, así que la nueva no se lee nunca.

Una keyColumn evita ambos compromisos: las filas del límite se excluyen en la cláusula WHERE, así que ni se releen ni se pierden. Es el ajuste recomendado exactamente por eso. Un límite exclusivo pierde en silencio las filas que comparten la marca de tiempo exacta y no se habían leído cuando terminó la ejecución. Uno inclusivo relee unas pocas filas, lo que no cuesta nada: cada escritura es un upsert idempotente, así que una fila que llega dos veces se guarda una.