Plantilla de Google Dataflow de Pub/Sub a ClickHouse

La plantilla de Pub/Sub a ClickHouse es un pipeline de streaming que lee mensajes codificados en JSON de una suscripción de Pub/Sub y los escribe en una tabla de ClickHouse. Los mensajes que no se pueden analizar o asignar al esquema de destino se envían a un destino dead-letter: una tabla de ClickHouse, un tema de Pub/Sub o ambos.

• La suscripción de Pub/Sub de origen debe existir.
• Los mensajes publicados en la suscripción deben ser JSON válidos.
• La tabla de ClickHouse de destino debe existir y sus nombres de columna deben coincidir con los nombres de los campos de la carga útil JSON.
• El host de ClickHouse debe ser accesible desde las máquinas worker de Dataflow.
• Se debe proporcionar al menos un destino dead-letter (clickHouseDeadLetterTable o deadLetterTopic). Si se proporcionan ambos, los mensajes fallidos se enrutan a ambos destinos simultáneamente.
• Cuando se establece clickHouseDeadLetterTable, la tabla dead-letter ya debe existir en ClickHouse con el esquema que se muestra en Gestión de mensajes dead-letter.
• Cuando se establece deadLetterTopic, el tema de Pub/Sub ya debe existir.

Los mensajes de Pub/Sub deben ser objetos JSON cuyos nombres de campo de nivel superior coincidan exactamente con los nombres de las columnas de la tabla de ClickHouse de destino.

Para mapear los mensajes entrantes a la tabla de destino, el pipeline realiza lo siguiente al iniciarse:

1. Obtiene el esquema de la tabla de ClickHouse de destino.
2. Crea un esquema Row de Beam a partir de ese esquema de ClickHouse.
3. Para cada mensaje entrante de Pub/Sub, analiza la carga útil JSON y arma una fila leyendo los campos indicados en el esquema de ClickHouse.

Los valores JSON se convierten al tipo de columna de ClickHouse correspondiente:

Como la pipeline es de streaming, las filas entrantes se acumulan en ventanas antes de escribirse en ClickHouse. La estrategia de ventanas se elige según los parámetros que proporciones:

Ajustar estos valores permite equilibrar la latencia y la eficiencia de inserción. Las ventanas más pequeñas reducen la latencia de extremo a extremo; las más grandes generan menos lotes INSERT, pero de mayor tamaño.

Los mensajes que fallen durante el análisis de JSON, el mapeo del esquema o la coerción de tipos se enrutan al destino o destinos dead-letter configurados. Debe proporcionarse al menos uno de clickHouseDeadLetterTable o deadLetterTopic; si ambos están configurados, los mensajes fallidos se envían a ambos.

Cuando se configura clickHouseDeadLetterTable, la tabla dead-letter ya debe existir con este esquema fijo:

Una definición mínima para una implementación de un solo nodo:

CREATE TABLE my_table_dead_letter (
    raw_message   String,
    error_message String,
    stack_trace   String,
    failed_at     DateTime
) ENGINE = MergeTree()
ORDER BY failed_at;

Cuando se establece deadLetterTopic, cada mensaje fallido se vuelve a publicar en el tema con:

• Carga útil: los bytes del mensaje original.
• Atributo errorMessage: el mensaje de excepción capturado en el momento del fallo.
• Atributo failedAt: la marca temporal de procesamiento en la que falló la fila.

Esto facilita reprocesar los mensajes fallidos una vez que se haya resuelto el problema subyacente del esquema o del productor.

La plantilla Pub/Sub to ClickHouse está disponible en Google Cloud Console.

Inicie sesión en Google Cloud Console y busque Dataflow.

1. Presione el botón CREATE JOB FROM TEMPLATE.

   

2. Una vez que se abra el formulario de la plantilla, introduzca un nombre para el job y seleccione la región deseada.

3. En el campo Dataflow Template, escriba ClickHouse o Pub/Sub y seleccione la plantilla Pub/Sub to ClickHouse.

4. Una vez seleccionada, el formulario se expande. Complete lo siguiente:

   • La suscripción de entrada de Pub/Sub, con el formato projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>.
   • La URL del endpoint de ClickHouse; para ClickHouse Cloud, use https://<HOST>:8443.
   • La base de datos de ClickHouse, la tabla de destino, el nombre de usuario y la contraseña.
   • Al menos un destino dead-letter: una tabla de ClickHouse o un tema de Pub/Sub (o ambos).

5. De forma opcional, personalice la agrupación en lotes (windowSeconds, batchRowCount) y los parámetros de ajuste de ClickHouseIO, como se detalla en la sección Parámetros de la plantilla.

Ve a la pestaña Dataflow Jobs en tu Google Cloud Console para supervisar el estado del trabajo. Allí encontrarás los detalles del trabajo, incluidos el progreso y cualquier error:

La plantilla también emite las siguientes métricas personalizadas en el espacio de nombres PubSubToClickHouse, visibles desde la página del trabajo de Dataflow:

Este error se produce cuando ClickHouse se queda sin memoria al procesar lotes grandes de datos. Para resolver este problema:

• Aumente los recursos de la instancia: actualice su servidor ClickHouse a una instancia más grande y con más memoria para soportar la carga de procesamiento de datos.
• Reduzca el tamaño del lote: reduzca batchRowCount (y/o maxInsertBlockSize) en la configuración del trabajo de Dataflow para enviar fragmentos de datos más pequeños a ClickHouse, lo que reduce el consumo de memoria por lote.

Las causas más comunes son:

• Los nombres de los campos JSON no coinciden exactamente con los nombres de las columnas de ClickHouse (la coincidencia es sensible a mayúsculas y minúsculas).
• El valor JSON no se puede convertir implícitamente al tipo de la columna (por ejemplo, una cadena que no siga ISO-8601 en una columna DateTime).
• El esquema de la tabla de destino ha cambiado desde que se inició el pipeline; el esquema se obtiene una sola vez al arrancar. Reinicie el job después de aplicar los cambios de esquema.

Inspeccione las columnas error_message y stack_trace de la tabla dead-letter de ClickHouse (o el atributo errorMessage en los mensajes dead-letter de Pub/Sub) para identificar la causa raíz.

• Confirma que la suscripción esté recibiendo mensajes: comprueba la métrica messages-received en la página del job de Dataflow.
• En el modo basado en tiempo (solo windowSeconds), las filas solo se vuelcan al finalizar cada ventana. Reduce windowSeconds para verificar que los volcados se estén produciendo.
• Verifica la conectividad de red entre los workers de Dataflow y el endpoint de ClickHouse (firewall, peering de VPC o private service connect).

El código fuente de la plantilla está disponible en:

• GoogleCloudPlatform/DataflowTemplates — el repositorio original de Google Cloud Platform.
• ClickHouse/DataflowTemplates — el fork de ClickHouse.