Experiencia de adopción de Flink SQL

• El equipo Azar Matching Dev Team de Hyperconnect eligió implementar el procesamiento de streaming con SQL, en lugar de código de aplicación, para dividir una app legacy monolítica de Flink que usaba 96 CPU
• Dividirla en varias apps de Flink mejoraba el aislamiento, pero aumentaba la carga operativa, así que concluyeron que Flink SQL era más adecuado en términos de productividad y eficiencia operativa, aprovechando el nivel de entendimiento interno de Flink del equipo
• Flink SQL ofrecía soporte para checkpoint/savepoint, HA de JobManager, redistribución de TaskManager, ventanas, joins, event time, watermark, UDF y conectores personalizados, por lo que encajaba mejor con la situación del equipo que ksqlDB o Spark Structured Streaming
• El entorno de producción se montó como un Session mode Flink Cluster sobre Kubernetes, y con la API de Flink SQL Gateway y GitHub Actions gestionaron el despliegue de queries y la detención de jobs bajo un enfoque GitOps
• Lleva cerca de un año funcionando de forma estable y en expansión, pero el redespliegue de queries y los cambios de configuración del clúster siguen siendo incómodos, por lo que planean mejorar esto con un enfoque basado en el patrón GitOps Controller

---

Contexto: por qué reemplazaron una app legacy de streaming pesada

• El Azar Matching Dev Team administraba varias aplicaciones basadas en Flink, y entre ellas había una app legacy pesada que consumía 96 CPU
• Esta app tenía una estructura monolítica que concentraba en un solo lugar funciones como join de múltiples eventos de matching, publicación condicional de eventos y almacenamiento de flags en Redis
• Después de cambiar los nodos de ejecución por un trabajo de infraestructura a nivel de toda la empresa, la app dejó de funcionar correctamente, y no era fácil resolverlo rápido solo con tuning
• Como la funcionalidad crítica de join de eventos ya estaba implementada en una nueva app de Flink de otro proyecto, necesitaban una forma de reemplazar la parte posterior al join, es decir, la publicación condicional de eventos y la ejecución de lógica

Comparación de alternativas

• Implementarlo como una sola app de Flink reducía los objetos a administrar, pero tenía alta probabilidad de volver a convertirse en una app enorme, y una falla en una parte podía afectar a otras funciones
• Dividirlo en varias apps de Flink permitía administrar cada app de forma independiente, pero a medida que aumentaba la cantidad de apps también crecía la carga de clúster, recursos y despliegue
• Flink SQL permitía definir la lógica con queries, desarrollar rápido y administrar un solo clúster, pero tenía la desventaja de que era difícil expresar lógica compleja únicamente con SQL y requería experiencia operando el clúster
• El equipo ya tenía un alto nivel de entendimiento sobre la implementación interna de Flink, y consideró que Flink SQL tenía ventajas en productividad y eficiencia operativa

Por qué eligieron Flink SQL

• Flink SQL permite implementar una app de procesamiento de eventos en streaming con SQL sin escribir directamente código de aplicación
• En términos de alta disponibilidad (HA), Flink soporta procesamiento con estado, y con checkpoint y savepoint puede guardar y restaurar el estado del trabajo de forma periódica o en el momento deseado
  • JobManager puede configurarse en modo HA con esquema leader-standby
  • Si falla parte de los TaskManager, según la estrategia de reintento del job se puede redistribuir el trabajo fallido a otros TaskManager
• Solo con sintaxis SQL se pueden manejar las funciones clave del procesamiento de streaming
  • Con SELECT se transforma la forma de los datos y con WHERE se filtran registros
  • Con JOIN se combinan varios streams y con UNION se pueden unir streams
  • Soporta procesamiento por ventanas como tumbling, hopping(sliding) y session window
  • Con event time y watermark se puede definir el rango permitido para datos tardíos
• Con UDF y conectores personalizados también se cubren requerimientos fuera de las funciones base
  • Gran parte del sistema legacy seguía patrones usando comandos Redis SET o INCR, y como no existía un conector oficial de Redis para Flink, escribieron su propio Redis Connector
  • En ese momento no había una función built-in para obtener la intersección de tipo ARRAY, así que la implementaron como UDF y la usaron en las queries

Comparación con ksqlDB y Spark Structured Streaming

• ksqlDB estaba incluido en la plataforma Confluent que la empresa ya usaba para Kafka, y también había casos de uso internos
• Sin embargo, consideraron que su comportamiento de HA en procesamiento stateful era ineficiente
  • En failover de operaciones con estado, hay que hacer replay completo del changelog donde se registran los cambios de state, lo que puede hacer que el failover tome mucho tiempo
  • El enfoque de mantener réplicas del stream de procesamiento y seguir actualizando el changelog al state interno implica ejecutar las mismas operaciones también en las réplicas, por lo que el consumo de recursos puede duplicarse
  • Más detalles en Configuring ksqlDB for High Availability | Confluent Developer
• Spark Structured Streaming es un motor de procesamiento de streaming basado en el motor Spark SQL
  • Había casos de uso internos y permitía escribir UDF y Custom Sink
  • Tiene un ecosistema más grande y más consolidado que Flink
• Spark opera en unidades de micro-batch, por lo que puede introducir latencia por registro, y en escenarios donde el procesamiento en tiempo real es importante puede ser menos favorable que Flink
• Como dentro del equipo casi no había experiencia con Spark y además necesitaban escribir un Custom Sink, no era fácil elegir Spark con confianza

Construcción del entorno de clúster

• En local, se puede descargar el binario desde la página oficial de Flink y ejecutar el clúster con {FLINK_HOME}/bin/start-cluster.sh
• Si se ejecuta {FLINK_HOME}/bin/sql-client.sh, se abre el CLI de Flink SQL y se pueden enviar queries de prueba como SELECT 1;
• Después de enviar una query, en la UI web de Flink se puede verificar que la query enviada fue convertida a un Job y está corriendo
• A finales de 2022 se liberó Flink SQL Gateway, permitiendo el envío de queries vía HTTP

Arquitectura operativa basada en Kubernetes

• Como la mayoría de los servicios internos corre sobre Kubernetes, el clúster de Flink SQL también se montó sobre Kubernetes
• Las apps existentes de Flink ya se desplegaban y operaban todas en Application mode
  • Es un enfoque donde se levanta un clúster separado para cada aplicación
  • En Kubernetes, cada app corre levantando su propio JobManager Pod y TaskManager Pod
  • Era ventajoso en términos de independencia y aislamiento entre apps, y para manejar configuración y dependencias por trabajo
• Como Flink SQL envía Jobs a un clúster ya levantado, fue necesario ejecutar JobManager y TaskManager en Session mode
• El clúster se construyó con base en la guía Stand Alone Cluster on Kubernetes
• Para el entorno HA se siguieron las configuraciones de High-Availability with Standalone Kubernetes, y para high-availability.storageDir se usó s3
• Consideraron que el enfoque Native Kubernetes no encajaba con su infraestructura interna, porque levanta el clúster mediante shell scripts provistos y ellos necesitaban definir y desplegar directamente la configuración de deployment

Configuración de HA e integración con S3

• Para HA e integración con S3 usaron la siguiente configuración en config.yaml

high-availability.type: kubernetes
high-availability.storageDir: s3://{s3-path-for-flinksql-recovery}
kubernetes.cluster-id: {cluster-id}
kubernetes.namespace: {k8s-namespace}


# Puede ser necesario otorgar permisos para poder acceder al clúster de Kubernetes a través de la service account dentro del namespace.
kubernetes.service-account: {k8s-service-account-for-flinksql}

• En un entorno HA se levantan dos pods de JobManager, y sus direcciones deben ser distintas para que la lógica de elección de líder y similares funcione correctamente
• Los argumentos de ejecución del contenedor de JobManager se configuraron así

args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]

• Con esta configuración, en el ConfigMap de Kubernetes se almacena la información del pod de JobManager elegido como líder actual y el Job ID actualmente en ejecución, entre otros datos usados para HA

Despliegue de queries con enfoque GitOps

• Todavía no existe una UI web ni una herramienta dedicada para Flink SQL provista directamente por Flink
• Evaluaron como PoC un caso de integración con Hue, pero en ese momento había problemas de compatibilidad de versiones con Flink SQL Gateway, se requería desarrollo adicional y también llevaba mucho tiempo preparar el entorno de desarrollo
• Como dentro de la empresa ya se usa mucho el patrón GitOps, implementaron GitHub Actions para desplegar queries o detener Jobs
• Dentro del repositorio crean carpetas por Job y guardan como archivos SQL las queries a ejecutar
• GitHub Actions recibe el nombre de la carpeta y determina el archivo SQL del cual extraer la query
• La implementación llama a la Flink SQL Gateway REST API y fue escrita en Python por ser simple y fácil de probar

Casos operativos y respuesta ante fallas

• No habían experimentado fallas en JobManager, pero con la configuración HA, si JobManager falla, otro JobManager puede ser elegido como líder y continuar el trabajo
• Los TaskManager sí fallaban ocasionalmente, y en la mayoría de los casos se trataba de pods reiniciados por políticas de QoS de Kubernetes
• Confirmaron que, incluso cuando fallaba parte de los TaskManager, el trabajo seguía al redistribuirse a otros TaskManager
• La mayoría de las fallas de queries se debía a entrada de datos anómalos o falta de recursos de cómputo
  • Al leer datos JSON, un formato JSON inválido puede ignorarse con la opción json.ignore-parse-errors
  • Cuando JSON_VALUE extrae datos de una ruta específica y ocurre un error porque no existe el valor o el tipo es distinto, se puede definir un valor por defecto con DEFAULT {VALUE} ON ERROR
  • Si el CPU de TaskManager supera el 100% o falta memoria, se vuelve a desplegar aumentando los recursos del TaskManager o el parallelism de la query
• Al reiniciar el clúster por cambios de configuración o por agregar UDF, en algunos casos fallaban ciertos Jobs
  • Muchas veces la causa era que la configuración de timeout o retry del Job no era adecuada
  • Ajustaron timeout y retry para que el Job no dejara de reintentar demasiado pronto y siguiera intentando hasta que el clúster quedara estable tras el reinicio

Restricciones al modificar queries y restaurar state

• Al modificar condiciones de una query y volver a desplegarla, los casos en que se puede restaurar el state con savepoint se limitan a cambios muy simples, como modificar valores en expresiones condicionales
• Si cambia una condición de ventana, también cambia el state, lo que dificulta mantener compatibilidad y puede impedir la restauración con savepoint
• Si es importante mantener el state pero los requerimientos cambian con frecuencia, puede ser mejor escribir directamente una app

Puntos de monitoreo

• Flink ofrece muchas métricas integradas, así que si se cuenta con la infraestructura interna de monitoreo y un Metric Reporter adecuado, el entorno de monitoreo se puede montar fácilmente
• numRunningJobs indica la cantidad de Jobs actualmente en ejecución en el clúster; si el valor cae de golpe y se mantiene así, se puede asumir que hay un Job fallido
• taskmanager.cpu.load y taskmanager.memory.used permiten entender el uso de recursos del clúster
• Con busyTimeMsPerSecond se puede ver qué tan ocupado está el TaskManager por Job
• Si Kafka se usa como source, records-lag-max permite verificar rápidamente el estado de latencia de los datos

Ejemplo: agregación por ventana de eventos de login en Kafka

• El ejemplo del apéndice recibe eventos desde Kafka y publica en Kafka la cantidad de eventos de login del último minuto cada 10 segundos
• Los datos de entrada están en formato JSON e incluyen los campos event_time, event_type y data.user_id
• La query configura pipeline.name, parallelism.default y table.exec.state.ttl
• La tabla de entrada login_event usa Kafka connector y formato JSON, y establece json.ignore-parse-errors en true
• row_time se genera a partir de event_time, y el watermark se configura para procesar eventos que lleguen con hasta 5 segundos de retraso respecto al máximo event time observado hasta el momento
• La tabla de salida windowed_login_count publica resultados a un topic de Kafka e incluye el campo proc_time AS PROCTIME()
• HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) crea una hopping window de 1 minuto con intervalo de 10 segundos, y COUNT(*) agrega la cantidad de registros dentro de la ventana

Resultados operativos y mejoras pendientes

• El equipo pudo aprovechar su experiencia previa con Flink para agregar varias funciones de manera más fácil y rápida que antes
• Evalúan que obtuvieron resultados satisfactorios en términos de productividad y eficiencia operativa
• Después de la adopción, ha funcionado de forma estable durante cerca de 1 año sin trabajo operativo relevante
• Actualmente lo están expandiendo gradualmente
• Siguen existiendo incomodidades en el redespliegue de queries y los cambios de configuración del clúster, y planean mejorar el entorno de despliegue de queries implementando el patrón GitOps Controller

[flgkselql98]

En sistemas distribuidos como Flink, parece que hay que mantener 2 o 3 racks para asegurar HA, y da la impresión de que al integrarlo con Kubernetes lograron garantizar esa alta disponibilidad. Pero al final también habría que pensar en los recursos de los nodos worker de Kubernetes, así que me hace preguntarme si armaron nodos dedicados solo para Flink (porque cuando Flink tenga carga, podría haber problemas de caída en los nodos worker).
Desde esa perspectiva, ¿realmente hay ventajas en usar Kubernetes?

Además, cuando usas funciones de ventana en Flink, esos datos se mantienen en memoria mientras funciona el JOIN en SQL, así que viéndolo desde el punto de vista del trade-off, me pregunto si Flink realmente es una buena opción. Si con el tiempo ese SQL + job que se vuelve cada vez más grande termina muriendo, el problema que se genera sería enorme.

Yo también me pregunto, en situaciones donde se necesita un JOIN en el data source de más alto nivel, cómo se podría bajar eso al nivel de la aplicación y procesarlo sin usar Flink.