Flink JOIN en tiempo real: procesamiento de actualización con delete+insert

# Manejo de actualizaciones en JOIN en tiempo real con Apache Flink: Resolviendo problemas de fugas de estado

En sistemas de JOIN en tiempo real de Apache Flink, las actualizaciones en los registros de la tabla fuente pueden provocar fugas de estado y resultados incorrectos. Si el campo user_id en la tabla domains cambia, el evento de actualización podría aterrizar en un TaskManager diferente sin limpiar los datos antiguos. De manera similar, cambiar la clave primaria genera duplicados downstream sin eliminar los registros obsoletos.

Desglosemos los escenarios:

• Cambiar user_id con un domains.id fijo: El nuevo evento se hashea a una clave diferente, dejando la antigua en el estado.
• Cambiar tanto domains.id como user_id: Se crea un nuevo registro, pero el anterior no se elimina, rompiendo la desduplicación.

Arquitectura distribuida de Flink y zonas locales

Flink distribuye los datos entre TaskManagers mediante keyBy. El hash de la clave determina el contenedor propietario, garantizando:

• Que todos los eventos de una clave se procesen localmente.
• Que se preserve el orden de los eventos.

El espacio de hash se distribuye de forma uniforme, con cada TaskManager manejando su zona. Sin keyBy, recurre a round-robin, pero esto no sirve para JOIN: pierde las garantías de localidad.

Solución universal: Eliminar + Insertar

Para manejar actualizaciones, dividimos los eventos CDC de Debezium en eliminar (antes) e insertar (después). El orden importa: eliminar por la clave antigua primero, luego insertar por la nueva. keyBy garantiza el procesamiento secuencial.

Implementación en la clase Domain:

public class Domain implements Serializable {
  public Integer id;
  public Integer user_id;
  public String domain_name;
  public boolean delete;

  // getters and setters omitidos

  public static Domain[] fromRow(Row row) {
        Character op = (Character) row.getField(1);

        if (op == null) {
            throw new IllegalStateException("Nunca debería pasar, si Debezium está bien");
        }

        if (op == 'd') {
            Row before = (Row) row.getField(0);
    
            return new Domain[]{
              new Domain(before.getField(0), before.getField(1), before.getField(2), true)            };
        } else if (op == 'u') {
            Row before = (Row) row.getField(0);
            Row after = (Row) row.getField(2);
    
            return new Domain[]{
              new Domain(before.getField(0), before.getField(1), before.getField(2), true),
              new Domain(after.getField(0), after.getField(1), after.getField(2), false)
            };
        } else {
            Row after = (Row) row.getField(2);
    
            return new Domain[]{
              new Domain(after.getField(0), after.getField(1), after.getField(2), false)
            };
        }
    }
}

Lógica:

• Eliminar ('d'): Genera un evento de eliminación del "before".
• Actualizar ('u'): Eliminar del "before" + insertar del "after".
• Insertar ('c'): Insertar solo del "after".

La eliminación marca el registro como removido en el estado (o lo elimina), mientras que la inserción lo actualiza. Para claves JOIN o PK mutables, el evento se autoclimpia el estado.

Beneficios de este enfoque

• Soporte completo de CRUD en flujos en tiempo real.
• Estado sin fugas ni redundancias.
• Manejo correcto de claves JOIN y PK mutables.
• Orden preservado mediante keyBy.

Lecciones clave

• Dividir actualizaciones en eliminar+insertar soluciona fugas de estado en cambios de clave.
• Eliminar antes de insertar es crucial para la consistencia.
• El "before/after" de Debezium da acceso a valores antiguos.
• Escala sin problemas en la configuración distribuida de Flink.

— Editorial Team