Documentation for a newer release is available. View Latest

Purging

Descripción general

El propósito de ODS Purging es eliminar todos los datos persistidos (UnitOfWork, Summary, MdsObjects, PdsObjects, ProcessObjects, CustomObjects) asociados con un determinado unit-of-work-id que sea más antiguo que un periodo configurado. ODS no está pensado como una solución de persistencia a largo plazo, por lo que será necesario eliminar datos pasado cierto punto para evitar implicaciones de almacenamiento masivo.

Hay dos modos de purgado, STANDARD que debe usarse cuando la base de datos subyacente es mongo y TTL cuando la base de datos subyacente es cosmos. La elección de uno de ellos se hace estableciendo el siguiente valor de configuración ods.purging.mode desde la configuration.

STANDARD Purging

Conceptos clave

La implementación de purgado de ODS utiliza un Akka Cluster Singleton y un Akka Scheduler para eliminar datos con frecuencia en pequeños lotes. Estas eliminaciones frecuentes se producirán periódicamente a lo largo del día, con un PurgeReport generado y actualizado a medida que se eliminan datos. El objetivo es eliminar todos los datos necesarios sin afectar a ODS-Ingestion.

Periodo de retención

El periodo de retención es una cantidad de tiempo basada en fecha, configurable, que se usa para determinar si los datos persistidos cumplen uno de los criterios de purga. Por defecto, se establece en 2 años; por ejemplo, si la purga se está ejecutando el 17/05/23, el periodo de retención sería: 17/05/21 - 17/05/23.

El límite inferior del periodo de retención es la fecha de ejecución de la purga, p. ej., 2023-05-17, menos el periodo de retención, p. ej., 2 años, y al inicio del día en UTC, resultando en 2021-05-17T00:00:00.000Z.

Se considera que un unitOfWork está fuera del periodo de retención cuando su campo finishedAt es anterior al límite inferior del periodo de retención.

Campos de UnitOfWork: startedAt, finishedAt, archivedAt

Los campos startedAt y finishedAt de unitOfWork se rellenan con marcas de tiempo de ciertos ProcessFlowEvents que son ingeridos por ODS. startedAt se mapea desde el primer ProcessFlowEvent ingerido para un unitOfWorkId. finishedAt se mapea desde el ProcessFlowEvent ingerido que indica que el unitOfWork ha alcanzado un estado global terminal. Si se despliega ipf-archiver-application, el campo archivedAt de unitOfWork se rellenará cuando un unitOfWork y todos sus objetos ODS relacionados hayan sido archivados.

Criterios de purga

Un unitOfWork será purgado si ocurrió fuera del periodo de retención; la fecha y hora que se utiliza para determinar esto depende de la configuración terminal-unit-of-works-only.

Si la configuración terminal-unit-of-works-only es true
- finishedAt del unitOfWork es anterior al límite inferior del periodo de retención
Si la configuración terminal-unit-of-works-only es false
- finishedAt del unitOfWork es anterior al límite inferior del periodo de retención
- SI finishedAt del unitOfWork no existe, entonces startedAt del unitOfWork es anterior al límite inferior del periodo de retención

IMPORTANTE: Si la configuración terminal-unit-of-works-only es false, la funcionalidad de purga incluye unitOfWorks no terminales.

Archived dependent journey types

La configuración archived-dependent-journey-types se utiliza para definir los tipos de Journey de unit of work que deben haber sido archivados antes de que puedan ser elegibles para el purgado. Por ejemplo, si se establece en ["PAYMENT"], los unit of works con journeyType == PAYMENT deben haber sido archivados antes de poder ser purgados.

Los tipos de journey que no estén definidos en esta lista pueden purgarse sin ser archivados, siguiendo los otros criterios de purga definidos arriba.

IMPORTANTE: Si no se definen archived-dependent-journey-types, todos los tipos de journey pasarán este criterio para el purgado.

Los siguientes ejemplos suponen una fecha de ejecución de 2023-05-17 y un periodo de retención de dos años, con un límite inferior de 2021-05-17.

Tabla 1. Ejemplos
UnitOfWork	Configured terminalUnitOfWorksOnly	Configured archivedDependentJourneyTypes	Purged?	Notes
started at: `2021-05-16` finished at: `2021-05-16`	`false`	[]	YES	El unitOfWork finalizó antes del límite inferior del periodo de retención
started at: `2021-05-17` finished at: `2021-05-17`	`false`	[]	NO	El unitOfWork finalizó dentro del límite inferior del periodo de retención
started at: `2021-05-16` finished at: `null`	`false`	[]	YES	El unitOfWork no ha finalizado y empezó antes del límite inferior del periodo de retención
started at: `2021-05-16` finished at: `2021-05-16`	`true`	[]	YES	El unitOfWork finalizó antes del límite inferior del periodo de retención
started at: `2021-05-17` finished at: `2021-05-17`	`true`	[]	NO	El unitOfWork finalizó dentro del límite inferior del periodo de retención
started at: `2021-05-16` finished at: `null`	`true`	[]	NO	El unitOfWork no ha finalizado y empezó antes del límite inferior del periodo de retención
started at: `2021-05-16` finished at: `2021-05-16` archivedAt: `2021-05-16` journeyType: `PAYMENT`	`true`	["PAYMENT"]	YES	El unitOfWork finalizó antes del límite inferior del periodo de retención y ha sido archivado
started at: `2021-05-16` finished at: `2021-05-16` archivedAt: `null` journeyType: `PAYMENT`	`true`	["PAYMENT"]	NO	El unitOfWork finalizó antes del periodo de retención pero no ha sido archivado
started at: `2021-05-16` finished at: `2021-05-16` archivedAt: `null` journeyType: `RECALL`	`true`	["PAYMENT"]	YES	El unitOfWork finalizó antes del límite inferior del periodo de retención. El tipo de journey RECALL no está especificado en la configuración; por lo tanto, no importa si este pago ha sido archivado o no

Ejecución recurrente

Para gestionar mejor la carga de la base de datos y el manejo de excepciones, la ejecución de la purga se divide en muchas purgas más pequeñas que ocurren periódicamente a lo largo del día. Al comienzo del día, se persistirá en la base de datos un PurgeReport. Luego, utilizando una frecuencia configurada, se producirán purgas recurrentes más pequeñas de un tamaño configurado. Estas purgas más pequeñas seguirán ejecutándose hasta que se hayan eliminado todos los datos necesarios.

Purge Report

Los detalles de una ejecución de purga se persisten como un PurgeReport. Cada día, se creará un nuevo PurgeReport y luego se actualizará durante la ejecución de la purga.

public final class PurgeReport {
    private LocalDate executionDate;
    private Period retentionPeriod;
    private Instant retentionPeriodLowerBound;
    private boolean terminalUnitOfWorksOnly;
    private List<String> archivedDependentJourneyTypes;
    private long summariesToDelete;
    private long summariesDeleted;
    private Instant startedAt;
    private Instant finishedAt;
    private Duration duration;
}

Tabla 2. Purge Report Overview
Field	Description	Example
executionDate	The date which this purge was executed. This will be unique for each PurgeReport	2021-05-17T00:00:00.000Z
retentionPeriod	The configured `retentionPeriod` for this purge execution	P2Y
retentionPeriodLowerBound	The date-time lower bound for the retention period. A unit-of-work-id passes one of the criteria to be purged if its `Summary.lastUpdated` date-time is less than the `retentionPeriodLowerBound`	2021-05-17T00:00:00.000Z
terminalUnitOfWorksOnly	The configured `terminal-unit-of-works-only` flag for this purge execution. If configured to `true` only unit-of-work-ids that have reached a terminal global status will be eligible for purging	false
archivedDependentJourneyTypes	The configured `archived-dependent-journey-types` for this purge execution. Unit of works with journey types defined in this config must be archived before being eligible for purging	[PAYMENT, RECALL]
unitOfWorksToDelete	The number of Summary documents that have been identified as outside the retention period and should be deleted during this purge execution	1234567
unitOfWorksDeleted	The number of Summary documents that have actually been deleted. This value is incremented as the purge is ongoing	1234567
startedAt	The date-time at which purging began for this execution date (will nearly always be the start of the day)	2023-05-17T00:00:02.170Z
finishedAt	The date-time at which all the data that should be deleted for a given execution date has been deleted	2023-05-17T00:32:03.180Z
duration	The duration of time taken to complete a purge. Calculated by comparing the `startedAt` and `finishedAt` fields	PT32M1.01S

Uso

ODS purging solo se realiza dentro de la aplicación ods-ingestion y está deshabilitado por defecto. Para habilitar el purgado, establece el siguiente valor de configuración: ods.purging.enabled = true.

Con esto habilitado, se configurará el Akka Cluster Singleton y la funcionalidad de purga comenzará a ejecutarse periódicamente de acuerdo con la configuración establecida.

Se puede encontrar más información sobre la configuración de ODS Purging abajo.

Implementación

Ods Persistence Purging Port

Cualquier implementación de purga debe proporcionar un Spring Bean para la interfaz PurgingOperations definida en ods-persistence-purging-port. Estos son métodos de base de datos que deben consultar, actualizar y eliminar datos del tipo de base de datos que se utilice.

MongoPurgingOperations

Actualmente, IPF-ODS admite MongoDB y Azure CosmosDB. Por lo tanto, la implementación predeterminada de la interfaz PurgingOperations es la clase MongoPurgingOperations. Esta utiliza la librería mongodb-reactivestreams-client para interactuar con las colecciones de ODS.

Purger

Cualquier implementación de purga debe proporcionar un Spring Bean para la interfaz Purger. Por defecto, se utiliza una instancia de la clase DefaultPurger. DefaultPurger utiliza el Bean MongoPurgingOperations y, cuando se dispara el método purge(), hace lo siguiente:

Purgar un único unit-of-work-id

Un único IPF-Flow generará muchos objetos ODS que se persisten en múltiples colecciones. Estos objetos serán: un único UnitOfWork, un único Summary y múltiples MdsObjects, PdsObjects, ProcessObjects y CustomObjects, todos los cuales están vinculados por un unit-of-work-id único generado por IPF. Para que un unit-of-work-id se considere purgado correctamente, todos los datos relacionados con ese unit-of-work-id deben eliminarse.

Para asegurarlo en ODS purging, el Summary, todos los MdsObjects, PdsObjects, ProcessObjects y CustomObjects para un unit-of-work-id dado se eliminan antes de eliminar su UnitOfWork.

Manejo de errores

Debido a la naturaleza de la implementación de purga, los errores con la eliminación se vuelven a intentar de forma natural en la siguiente ejecución recurrente. Sin embargo, el PurgeReport persistido debe mantenerse lo más actualizado posible. Por lo tanto, se han implementado escrituras reintento-habilitadas para cualquier operación de base de datos que escriba en la colección purgeReports.

Cualquier error lanzado dentro de la ejecución de la purga se registra como una advertencia sin que se tome ninguna acción. La purga se ejecutará de nuevo en breve y los datos que no se pudieron purgar previamente se recogerán en ejecuciones posteriores.

Akka Cluster Singleton

Se utiliza un Cluster Singleton para ejecutar la purga recurrente. La clase PurgingSchedulerSingleton se registra como un Spring Bean al inicio. Esto configura un Singleton Actor con un Akka Scheduler.

Este Akka Scheduler toma una instancia de la interfaz Purger y dispara una purga, llamando al método Purger.purge(), a una tasa definida por la configured frequency. Por defecto, esta frecuencia está configurada para dispararse cada 1 segundo.

Throughput de borrado

El throughput de borrado se define configurando la frecuencia y el fetch size. Si la frecuencia es 1 segundo y el fetch size es 500, entonces ODS intentará eliminar 500 unitOfWorks cada segundo.

Eliminar 500 unitOfWorks por segundo puede no ser alcanzable, dependiendo de los recursos de base de datos disponibles y de la carga de ingestión en el momento del borrado. El borrado también tendrá un impacto (a veces significativo) en la ingestión.

Cada ejecución (cada segundo) encuentra 500 unitOfWorks candidatos, los agrupa en lotes de ~62 (asumiendo un paralelismo de 8) y realiza 4 operaciones de borrado, pasando a cada una la lista de 62 unit of work ids. El borrado del summary, process, MDS, PDS y custom se hace de forma concurrente y, cuando estos han finalizado, se borran los unitOfWorks.

La latencia total de borrado es la suma de la mayor latencia de borrado entre summary, process, MDS, PDS y custom, más la latencia de borrado para unitOfWorks.

Posibles enfoques

Hay varios enfoques, con diferentes implicaciones.

Borrar tantos unitOfWorks candidatos como sea posible al comienzo del día

El objetivo es borrar todos los unitOfWorks candidatos antes de que se produzca cualquier carga de ingestión.

Dado un fetch size de 500 y una frecuencia de 1, ODS podría teóricamente borrar 5.4 millones de unitOfWorks candidatos entre las 12am y las 3am, cuando la carga de ingestión sería baja o nula.

Si hay alguna carga de ingestión significativa durante este periodo, el impacto en el throughput de ingestión también será significativo, al igual que el impacto en el throughput de borrado.

Este enfoque se basa en que las operaciones de borrado tengan una latencia suficientemente baja, de modo que todos los datos para los 500 unitOfWorks puedan borrarse en 1 segundo.

Si la latencia total de borrado (la mayor latencia de borrado entre summary, process, MDS, PDS y custom, más la latencia para borrar los unitOfWorks) supera 1 segundo, las ejecuciones posteriores se irán encolando.

Tabla 3. Ejemplos
Deletion Throughput/s	Time to delete 1 million units of work (h:m:s)
100	2:46:40
200	1:23:20
500	0:33:20

Borrar un pequeño número de unitOfWorks candidatos durante todo el día

El objetivo es borrar lentamente durante todo el día, evitando el impacto en el throughput de ingestión.

Dado un fetch size de 40 y una frecuencia de 1, ODS podría teóricamente borrar 3.4 millones de unitOfWorks candidatos durante todo el día.

ODS está siempre borrando (asumiendo que quedan candidatos), incluso cuando la carga de ingestión es alta.

Estos borrados frecuentes pero pequeños pueden seguir afectando a una ingestión de alto throughput.

Tabla 4. Ejemplos
Deletion Throughput/s	Deleted in 24 Hours
15	1,296,000
20	1,728,000
40	3,456,000

Enfoque preferido

El enfoque preferido, y actualmente por defecto, es borrar pequeños lotes de unitOfWorks candidatos con frecuencia, durante todo el día.

En pruebas contra CosmosDB, no se pudo lograr un throughput de borrado alto con la implementación actual, debido a la alta latencia de las operaciones de borrado. Borrar a 40 tps introdujo lag de ingestión con solo 300 tps de throughput de ingestión, y a 500 tps, el lag de ingestión se disparó y continuó aumentando. Para despliegues con CosmosDB se recomienda un fetch size mucho menor, algo como 16.

En pruebas contra MongoDB, se alcanzó un throughput de borrado de alrededor de 1000 tps, pero para evitar cualquier impacto significativo en la ingestión, se recomienda un throughput de borrado de 500 tps. Usar un número más pequeño, p. ej., 40 tps, probablemente sea preferible para distribuir la carga de borrado a lo largo del día.

En última instancia, dependerá de los recursos de base de datos disponibles y del requisito del cliente. Algo intermedio entre los enfoques anteriores podría ser más adecuado.

TTL Purging

La funcionalidad de time-to-live (TTL) permite que la base de datos expire datos automáticamente. El modo de purgado TTL debe usarse cuando se despliega la base de datos cosmos.

Establecer el valor de time to live para una colección

Para habilitar TTL de forma universal en una colección, es necesario crear un "TTL index" (índice time-to-live). El índice TTL es un índice sobre el campo _ts con un valor expireAfterSeconds. Una vez que se crea el índice, la base de datos eliminará automáticamente cualquier documento en esa colección que no haya sido modificado en los últimos expireAfterSeconds segundos.

Establecer el valor de time to live para un documento

También se admiten valores TTL por documento. El/los documento(s) deben contener una propiedad de nivel raíz "ttl" (en minúsculas) y debe haberse creado un índice TTL como el descrito arriba para esa colección. Los valores TTL establecidos en un documento sustituirán el valor TTL de la colección.

Hay algunas funciones de mantenimiento implementadas en el modo de purgado TTL y se puede encontrar más información aquí TTL Purging Housekeeping jobs.

Configuración predeterminada

ods {
  purging {
    enabled = false
    mode = STANDARD
    retention-period = 2Y
    terminal-unit-of-works-only = false
    archived-dependent-journey-types = []

    frequency = 1s
    fetch-size = 16
    parallelism = 8
  }
}

Tabla 5. Resumen de configuración de Purge
Config Key	Description
`ods.purging.enabled`	Indicador para habilitar o deshabilitar ODS purging. `true` o `false`
`ods.purging.mode`	Modo de purgado. `STANDARD` o `TTL`
`ods.purging.retention-period`	El periodo de retención para la ejecución de purga. Cantidad de tiempo basada en fecha, p. ej., 1Y = 1 año, 1M = 1 mes, 1W = 1 semana
`ods.purging.terminal-unit-of-works-only`	Indicador para habilitar al purger a que solo purgue unitOfWorks que hayan alcanzado un Terminal Global Status. `true` o `false`.
`ods.purging.archived-dependent-journey-types`	Lista utilizada para definir los tipos de journey de unitOfWork que deben archivarse antes de ser elegibles para purgar. Opciones de journey type: `PAYMENT`, `RECALL`, `BULK` y `BATCH`. Por defecto, todos los tipos de journey, archivados o no, pasarán este criterio para el purgado.
`ods.purging.frequency`	La frecuencia a la que se producen ejecuciones recurrentes. Cantidad de tiempo basada en tiempo, p. ej., 2S = 2 segundos, 1M = 1 minuto
`ods.purging.fetch-size`	Número de unitOfWorks que se eliminarán en una ejecución
`ods.purging.parallelism`	Número de lotes de unitOfWorks que se eliminarán concurrentemente. Si el `fetch-size` es `40` y el `parallelism` es `8`, habrá 8 lotes de 5 unitOfWorks para eliminar en una ejecución,

ods.purging.fetch-size y ods.purging.frequency se utilizan juntos para determinar el número de unitOfWorks a eliminar y con qué frecuencia. Si ods.purging.fetch-size = 1000 y ods.purging.frequency = 1s, ODS intentará eliminar 1000 unitOfWorks (y todos los objetos ODS relacionados) por segundo.