Streaming de datos

El streaming de datos reenvía eventos de Joryio a destinos externos casi en tiempo real. Cada flujo vincula un destino a un conjunto de eventos y Joryio entrega por lotes los eventos que coinciden con esa selección.

Ve a Datos → Streaming de datos para gestionar los flujos.

Destinos

Destino	Tipo	Función
Webhook	Endpoint HTTP	Envía cada evento como JSON mediante POST a cualquier URL HTTPS
Snowflake	Almacén de datos	Carga eventos continuamente en una tabla de Snowflake
BigQuery	Almacén de datos	Transmite eventos a un conjunto de datos de BigQuery
Amazon Redshift	Almacén de datos	Carga eventos en un clúster de Amazon Redshift
Amazon S3	Almacenamiento de objetos	Escribe archivos JSON / NDJSON por lotes en un bucket
Kafka	Flujo de eventos	Publica eventos en un tema de Kafka
Mixpanel	Analítica de producto	Reenvía eventos a un proyecto de Mixpanel
Amplitude	Analítica de producto	Reenvía eventos a un proyecto de Amplitude

Campos de conexión

Cada destino solicita sus propios datos de conexión. Los secretos - contraseñas, claves, tokens, JSON de cuentas de servicio y valores de encabezados del webhook - se cifran cuando se almacenan y nunca se vuelven a mostrar; consulta Credenciales con privilegios mínimos.

• Webhook: URL, secreto de firma (opcional), encabezados adicionales (opcional, un objeto JSON como {"Authorization": "Bearer <token>"}; sus valores se tratan como secretos).
• Snowflake: cuenta, usuario, contraseña, rol (opcional: el rol con el que se ejecutan las instrucciones; si está vacío, se usa el rol predeterminado del usuario), almacén, base de datos, esquema y tabla.
• BigQuery: ID de proyecto, ID de conjunto de datos, ID de tabla y JSON de cuenta de servicio.
• Amazon Redshift: host, puerto, base de datos, usuario, contraseña, esquema y tabla.
• Amazon S3: bucket, prefijo, región, endpoint (opcional, para almacenamiento compatible con S3), ID de clave de acceso y clave de acceso secreta.
• Kafka: brokers, tema, usuario/contraseña SASL y SSL.
• Mixpanel: token de proyecto y región (EE. UU. / UE / India).
• Amplitude: clave API y región (EE. UU. / UE).

Credenciales con privilegios mínimos

Da a cada flujo su propia credencial, capaz de hacer una sola cosa: escribir en la tabla, la ruta del bucket o el tema al que entrega el flujo. Si la credencial llegara a filtrarse, eso es todo lo que se puede hacer con ella - no puede leer tus datos ni tocar nada más. Los ejemplos usan nombres de ejemplo (example-bucket, analytics.joryio_events, joryio-events); sustitúyelos por los tuyos.

Cómo trata Joryio las credenciales guardadas

• Cifradas. Las contraseñas, claves, tokens, JSON de cuentas de servicio y valores de encabezados del webhook se cifran antes de guardarse.
• Nunca se vuelven a mostrar. El editor y la API muestran un secreto guardado como ********. Nadie - ni tú ni el personal de Joryio - puede volver a leerlo en el panel ni en la API.
• Se conservan al editar otros campos. Un campo secreto que dejas vacío (o como ********) conserva el valor guardado; solo un valor que escribas lo reemplaza. Cambiar el nombre de un flujo o sus eventos nunca toca sus credenciales.
• Se eliminan con el flujo. Al eliminar un flujo se eliminan sus credenciales guardadas. Cuando se elimina la cuenta, se purgan con el resto de los datos de la cuenta.

Cuando elimines un flujo, revoca o rota también su credencial en el destino: Joryio ya no la tiene, pero la clave sigue siendo válida hasta que la revoques.

Los encabezados del webhook también son secretos. Introdúcelos como un objeto JSON, por ejemplo {"Authorization": "Bearer <token>"}. Los nombres de los encabezados siguen visibles; cada valor se cifra y se muestra como ********. Al editar el flujo, deja ******** para conservar un valor, escribe uno nuevo para reemplazarlo o quita un encabezado para eliminarlo. Usa un token emitido solo para Joryio, para poder revocarlo sin afectar a nada más.

Amazon S3

Crea un usuario de IAM (o un rol) para Joryio cuyo único permiso sea s3:PutObject en el bucket y el prefijo del flujo, y usa su clave de acceso en el flujo. Joryio escribe objetos bajo <prefix>/<date>/, así que con el prefijo joryio-events:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "JoryioWriteEvents",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::example-bucket/joryio-events/*"
    }
  ]
}

Probar conexión comprueba que el bucket existe, lo que AWS solo permite con s3:ListBucket en el bucket. Ese permiso deja a la clave listar los nombres de los objetos del bucket, así que es opcional: sin él, la entrega funciona y solo la prueba informa de un fallo. Para permitir la prueba, añade esta instrucción:

{
  "Sid": "JoryioTestConnection",
  "Effect": "Allow",
  "Action": "s3:ListBucket",
  "Resource": "arn:aws:s3:::example-bucket"
}

Para almacenamiento compatible con S3, crea un token limitado al bucket con permiso solo de escritura de objetos.

Amazon Redshift

Crea un usuario de base de datos dedicado que solo pueda insertar en la tabla del flujo:

CREATE USER joryio_loader PASSWORD '<strong-password>';
GRANT USAGE ON SCHEMA analytics TO joryio_loader;
GRANT INSERT ON TABLE analytics.joryio_events TO joryio_loader;

Introduce joryio_loader como usuario del flujo.

Snowflake

Crea un rol que solo pueda insertar en la tabla del flujo (más el USAGE que Snowflake necesita para llegar a ella) y un usuario con ese rol:

CREATE ROLE JORYIO_WRITER;
GRANT USAGE ON WAREHOUSE LOAD_WH TO ROLE JORYIO_WRITER;
GRANT USAGE ON DATABASE ANALYTICS TO ROLE JORYIO_WRITER;
GRANT USAGE ON SCHEMA ANALYTICS.EVENTS TO ROLE JORYIO_WRITER;
GRANT INSERT ON TABLE ANALYTICS.EVENTS.JORYIO_EVENTS TO ROLE JORYIO_WRITER;

CREATE USER JORYIO_LOADER PASSWORD = '<strong-password>'
  DEFAULT_ROLE = JORYIO_WRITER DEFAULT_WAREHOUSE = LOAD_WH;
GRANT ROLE JORYIO_WRITER TO USER JORYIO_LOADER;

Introduce JORYIO_LOADER como usuario y JORYIO_WRITER como rol. El rol es opcional: si lo indicas, Joryio ejecuta cada instrucción con ese rol; si lo dejas vacío, se aplica el rol predeterminado del usuario.

BigQuery

Crea una cuenta de servicio para Joryio y concédele BigQuery Data Editor (roles/bigquery.dataEditor) solo en el conjunto de datos del flujo - en la configuración de uso compartido del conjunto de datos, no en el proyecto. Después crea una clave JSON y pégala en JSON de cuenta de servicio.

Joryio envía filas con la API de inserción de BigQuery y lee los metadatos de la tabla para probar la conexión; ninguna de las dos ejecuta un trabajo de consulta, así que BigQuery Job User no es necesario. Data Editor en un conjunto de datos también puede modificar otras tablas del mismo, así que mantén la tabla del flujo en un conjunto de datos propio o concede un rol personalizado con solo bigquery.tables.get y bigquery.tables.updateData en la tabla.

Kafka

Crea un usuario SASL para Joryio (Joryio se autentica con SASL/PLAIN) y permítele solo WRITE y DESCRIBE en el tema del flujo. Crea el tema tú mismo de antemano - Joryio no necesita permiso para crear temas.

kafka-acls --bootstrap-server <broker:port> --command-config admin.properties \
  --add --allow-principal User:joryio \
  --operation Write --operation Describe \
  --topic joryio-events

Mixpanel, Amplitude y webhooks

• Mixpanel: el token de proyecto solo puede enviar eventos a su proyecto; no puede leer datos. Usa el token del proyecto al que debe alimentar el flujo.
• Amplitude: la clave API es la clave de ingesta de un proyecto. Usa el proyecto al que debe alimentar el flujo.
• Webhook: usa un secreto de firma y un token Authorization emitidos solo para Joryio, y acepta únicamente lo que envía el flujo.

Crear un flujo

1. Ve a Datos → Streaming de datos y haz clic en Crear flujo.
2. Asigna un nombre al flujo - se mostrará en la lista - y elige un destino.
3. Completa los campos de conexión del destino.
4. Elige el formato de carga útil: JSON (un objeto por evento) o NDJSON (una línea por evento).
5. Selecciona los eventos que quieres transmitir (ver abajo).
6. Revisa la barra de resumen y haz clic en Crear flujo.

Elegir eventos

Tú decides exactamente qué eventos recibe un flujo:

• All events - actívalo para transmitir todos los eventos actuales y futuros. Desactívalo para elegir eventos concretos.
• Mensajes, por canal - los eventos de mensajes se eligen por canal (consulta Selección por canal).
• Otros grupos - In-App, Journeys, Campañas, Wallet Passes, E-Commerce e Identidad. Marca un grupo completo o eventos sueltos.
• Eventos personalizados - en Custom Event, busca o escribe tus propios eventos (los que envías con track(), como purchase o page_view). La lista muestra los eventos que tu espacio de trabajo ha registrado; también puedes escribir uno que aún no se haya registrado y pulsar Add.

Selección por canal

El editor muestra una sección por canal - Email, SMS, WhatsApp, Push, Viber y Webhook - con solo los eventos que existen en ese canal (no hay "apertura de SMS"). Dentro de cada sección los eventos se agrupan:

Grupo	Opciones
Interacción	Envío, Entrega, Apertura (email), Lectura (WhatsApp), Clic
Problemas	Rebote, Marcado como spam, Error, Descartado, Aplazamiento, Respuesta automática (email)
Consentimiento	Baja, Alta
Entrantes	Recibido (SMS, WhatsApp, Viber)

Cada sección tiene su propia casilla de seleccionar todo, y Seleccionar todo en todos los canales lo marca todo a la vez. Junto a cada opción aparece un volumen mensual estimado - cuántos de esos eventos registró tu espacio de trabajo en los últimos 30 días - para que veas antes de guardar lo que costará la selección en tu destino.

Envío, Entrega y Apertura llevan la marca Alto volumen: cada uno es una fila por mensaje. Un flujo nuevo empieza con todas las demás opciones marcadas y deja estas tres para que las añadas tú.

El formato de los datos no cambia. Marcar Entrega en SMS transmite el evento estándar message.delivered cuya propiedad channel es sms - el mismo nombre y los mismos valores que usan el almacén de eventos, la API REST y las integraciones por webhook. Las respuestas entrantes conservan sus nombres: sms.message.received, whatsapp.message.received y viber.message.received.

Un flujo creado mediante la API todavía puede elegir un evento de mensaje por su nombre ("events": ["message.delivered"], opcionalmente acotado con "channels"). Eso sigue funcionando, y el editor lo muestra como el mismo evento marcado en cada canal que cubre.

Cambio de estado de suscripción

La sección Suscripciones tiene una opción, Cambio de estado de suscripción: cada alta y baja en cada canal y lista. Transmite los dos eventos de consentimiento que Joryio ya registra, así que nada se cuenta dos veces:

• message.subscribed - una persona se suscribió a un canal o a una lista. status y previousStatus indican el cambio, scope es channel o list (con listId) y source indica cómo ocurrió (api, dashboard, preference_center, double_opt_in, keyword, un alta en una tienda o un formulario, etc.).
• message.unsubscribed - una persona se dio de baja. scope es channel, list o global, reason indica cómo llegó (link, keyword, ...) y unsubscribeSource nombra la vía exacta (el botón de baja de la app de correo, el botón de la página, el centro de preferencias, una respuesta STOP).

Los rebotes y las quejas de spam son eventos propios (message.bounced, message.spamreport). Las importaciones de contactos, las sincronizaciones desde un almacén de datos y las sincronizaciones iniciales de tiendas escriben el consentimiento de forma masiva y no emiten estos eventos.

Recibos

Los informes de entrega, los rebotes, las quejas de spam, las bajas y los mensajes que Joryio descartó antes de enviarlos son recibos (receipts). Se eligen como cualquier otro evento de mensaje y llegan como estos eventos:

Evento	Canales	Propiedades destacadas
message.delivered	Email, SMS, Viber, Push, WhatsApp	messageId, campaignId, canvasId
message.bounced	Email	bounceType (hard o soft), bounceSubType, dsnStatus, reason
message.spamreport	Email	messageId, campaignId
message.failed	Todos los canales	errorCode, errorReason
message.skipped	Todos los canales	reason - por qué Joryio no lo envió (supresión, límite de frecuencia, consentimiento, horas de silencio)
message.deferred	Email	el proveedor reintenta tras un fallo temporal
message.auto_replied	Email	una respuesta automática de ausencia - el email se entregó
message.unsubscribed	Todos los canales	consulta Cambio de estado de suscripción
message.opened (canal whatsapp)	WhatsApp	la confirmación de lectura
sms.message.received, whatsapp.message.received, viber.message.received	SMS, WhatsApp, Viber	una respuesta entrante; en SMS y Viber, keyword es stop, start o help cuando la respuesta fue una palabra clave de cumplimiento

Los recibos llegan solo a Data Streaming. Nunca inician un journey ni activan una campaña, igual que antes.

Flujos creados antes de que existieran los recibos

Un flujo guardado antes de que existieran los recibos puede ya seleccionar algunos (por ejemplo message.bounced). Nunca llegaron y no empiezan a llegar por sí solos: el flujo muestra Los recibos ya están disponibles, y empiezan a fluir en cuanto vuelves a guardar el flujo. Los flujos nuevos reciben recibos desde el principio.

Qué contiene un evento de mensaje transmitido

• Cada evento lleva el id de Joryio del contacto (userId), y los eventos de mensajes llevan también el externalId del contacto cuando lo tiene.
• Las direcciones de email y los números de teléfono se incluyen solo cuando quien guardó el flujo por última vez tiene el permiso pii:view (administradores y propietarios siempre lo tienen). Si no, se eliminan. Volver a guardar el flujo aplica la regla a quien lo guarda.
• El contenido del mensaje (cuerpo, título, HTML) nunca se incluye en un recibo.
• Las aperturas y los clics que parecen automáticos (escáneres de seguridad, proxies de privacidad del correo) no se descartan; llevan $is_bot: true, para que decidas qué hacer con ellos.

Otros grupos de eventos

Grupo	Eventos
In-App	in_app.displayed, in_app.clicked, in_app.dismissed, in_app.converted
Journeys	canvas.experiment.entered, canvas.ai_decision.entered, canvas.step.entered, canvas.step.left
Campañas	campaign.sent, campaign.abtest.winner_declared, campaign.ai_decision.entered
Wallet Passes	pass.issued, pass.installed, pass.updated, pass.update_pushed, pass.removed, pass.voided, pass.expired, pass.scanned
E-Commerce	Order Completed, Order Refunded, Order Cancelled
Identidad	$identify, $alias

También puedes transmitir cualquier evento personalizado que envíes con track().

Procesamiento por lotes

Los eventos se almacenan temporalmente y se entregan por lotes. Puedes definir:

• Tamaño del lote: cuántos eventos se acumulan antes de enviarlo.
• Intervalo de vaciado: el tiempo máximo que se espera antes de enviar un lote parcial.

Un lote se entrega al alcanzar su tamaño configurado o cuando transcurre el intervalo de vaciado, lo que ocurra primero.

Cómo funciona

Los eventos llegan a los flujos mediante un bus de distribución duradero, independiente del almacenamiento de analítica. Cada flujo ve el flujo completo de eventos de todos los canales; por tanto, un destino sigue recibiendo eventos sin depender de cómo se escriba la analítica.

Cuando un evento coincide con la selección de un flujo, se pone en cola y se entrega al destino en el siguiente lote. Cada entrega incluye el nombre del evento, las propiedades, la marca de tiempo y el identificador de usuario; en los eventos de mensajería, también el channel de origen. Los almacenes y destinos de almacenamiento cargan el lote como filas o archivos; los destinos de analítica de producto reenvían cada evento mediante su API de ingesta; los webhooks reciben un cuerpo JSON o NDJSON.

Cada flujo agrupa eventos según el tamaño de lote y el intervalo de vaciado, y reintenta los envíos fallidos. Si un destino no está disponible, los eventos sin entregar vuelven a la cola y se reenvían cuando se recupera, para que una interrupción no provoque pérdida de datos.

Garantías de entrega y duplicados

La entrega es al menos una vez (at-least-once). En casos poco frecuentes - un reintento tras un fallo de red, o el reinicio de un proceso a mitad de un lote - el mismo evento puede entregarse a tu destino más de una vez.

Es algo deliberado, y es la misma garantía que ofrece cualquier plataforma de streaming. La alternativa, como mucho una vez, significaría descartar un evento en silencio cuando el resultado de la entrega es desconocido; preferimos enviarlo dos veces a perderlo.

Cada evento lleva un eventId estable. Es el mismo valor en cada entrega de ese evento, así que puedes usarlo para eliminar duplicados:

Destino	Qué hacemos
Amplitude	se envía como insert_id - Amplitude elimina duplicados automáticamente
Mixpanel	se envía como $insert_id - Mixpanel elimina duplicados automáticamente
S3	un lote reenviado se escribe en la misma clave de objeto, sobrescribiendo en lugar de añadir un segundo archivo
BigQuery, Snowflake, Redshift	elimina duplicados por eventId al cargar, p. ej. ROW_NUMBER() OVER (PARTITION BY eventId)
Webhook	ignora un eventId que ya hayas procesado

Para almacenes de datos y almacenamiento, los duplicados son inofensivos si los eliminas al cargar. Para webhooks que actúan sobre cada evento - crear un ticket, cobrar algo - comprueba el eventId antes de actuar.

Supervisión

La lista de flujos muestra los KPI y el estado de cada flujo:

• Activo: entrega con normalidad.
• En pausa: detenido temporalmente; puedes reanudarlo cuando quieras.
• Error: las entregas recientes están fallando; revisa la conexión.

También puedes ver el total entregado, el total fallido, la tasa de entrega y la hora de la última entrega. Abre un flujo para ver sus detalles, editarlo, pausarlo, reanudarlo o eliminarlo.