Procesar registros incrementales usando una marca de agua en Jitterbit Studio

Introducción

Una marca de agua es un valor almacenado que marca el punto más reciente que una operación de sincronización ha procesado. En cada ejecución, la operación lee la marca de agua almacenada, la utiliza para filtrar registros ya procesados, y actualiza la marca de agua después de completar el lote. Solo se recuperan registros nuevos o modificados en ejecuciones posteriores.

Esta guía demuestra el patrón usando casos de Salesforce como origen. La marca de agua es el máximo LastModifiedDate de la ejecución anterior. El patrón aplica a cualquier origen que exponga una marca de tiempo de modificación confiable y admita consultas filtradas.

Dos funciones de caché gestionan el valor almacenado:

• ReadCache: Recupera la marca de agua almacenada al inicio de cada ejecución.
• WriteCache: Actualiza la marca de agua después de completar el procesamiento.

Para una introducción a ReadCache y WriteCache, incluyendo opciones de alcance de caché y límites de velocidad, consulta Detectar y desduplicar registros usando funciones hash.

Patrón de diseño

Los pasos de lectura y actualización de la marca de agua enmarcan la lógica principal de consulta y procesamiento. La fecha almacenada avanza después de cada ejecución exitosa, por lo que la ventana de consulta se desplaza automáticamente hacia adelante.

Parte 1: Leer la marca de agua

Añade un paso de script como el primer paso de la operación. El script lee la marca de agua almacenada y utiliza una fecha predeterminada como respaldo en la primera ejecución:

// Set the cache key and expiration
cacheKey = $project_name + "_LastModifiedDate";
cacheExpiry = 2592000; // 30 days in seconds

// Read the stored watermark
watermarkDate = ReadCache(cacheKey, cacheExpiry, "project");

// Fall back to the default date if no value is stored
if(length(trim(watermarkDate)) == 0,
    watermarkDate = $default_watermark_date;
);

Clave de caché: Utiliza una clave única para este conjunto de datos dentro del proyecto. El uso de prefijos con el nombre del proyecto o un identificador de conjunto de datos (por ejemplo, $project_name + "_SF_Case_LM") evita colisiones de claves cuando múltiples marcas de agua se almacenan en el mismo proyecto.

Fecha predeterminada: default_watermark_date es una variable de proyecto establecida en Variables de proyecto a una fecha pasada lo suficientemente lejana para incluir todos los registros que deseas en la primera ejecución. Utiliza el formato ISO 8601 que Salesforce espera en SOQL: por ejemplo, 2000-01-01T00:00:00.000Z.

Expiración: 30 días (2592000 segundos) mantiene la marca de agua disponible en ejecuciones programadas. Aumenta este valor para operaciones que se ejecutan con menos frecuencia.

Alcance: El alcance "project" hace que el valor en caché sea accesible a todas las operaciones en el proyecto y lo persiste entre ejecuciones. Utiliza "env" si la marca de agua debe compartirse entre múltiples proyectos en el mismo entorno.

Parte 2: Consultar usando la marca de agua

Después de que el script lee la marca de agua, consulta el origen usando watermarkDate como límite de filtro.

Usar SfLookupAll en un script

SfLookupAll devuelve un arreglo bidimensional de registros coincidentes. Úsalo cuando el conjunto de resultados se pasa a un script o transformación posterior para procesamiento:

soql = "SELECT Id, LastModifiedDate FROM Case"
    + " WHERE LastModifiedDate >= " + watermarkDate
    + " AND AccountId != null"
    + " ORDER BY LastModifiedDate ASC";

$caseIds = SfLookupAll("<TAG>endpoint:salesforce/Salesforce</TAG>", soql);

Para más información sobre cómo construir y ejecutar consultas SOQL, consulta Consultar registros de Salesforce usando SOQL.

Usar una actividad Salesforce Query

Si utilizas una actividad Query de Salesforce en lugar de SfLookupAll, pasa watermarkDate como variable de Jitterbit y referencíala en el campo de condición de la actividad. Establece el operador de condición a mayor que o igual a e ingresa [watermarkDate] como valor.

Parte 3: Actualiza la marca de agua

Después de procesar todos los registros del lote, establece la marca de agua al máximo LastModifiedDate entre los registros realmente obtenidos en la Parte 2, luego escríbela en caché. Agrega esto como el último paso de script en la operación, o en la operación final de la cadena después de que se completen todas las operaciones secundarias:

// Derive the new watermark from the records retrieved in Part 2.
recordCount = Length($caseIds);
if(recordCount > 0,
    // Part 2 ordered results by LastModifiedDate ASC, so the last row holds the max.
    newWatermark = $caseIds[recordCount - 1]["LastModifiedDate"];
    WriteCache(cacheKey, newWatermark, cacheExpiry, "project");
);

Si utilizas la ruta de actividad Query de Salesforce de la Parte 2 en lugar de SfLookupAll, captura el máximo LastModifiedDate conforme se procesan los registros (por ejemplo, acumula el máximo en una variable global en la transformación de procesamiento), luego escribe ese valor en caché aquí.

El guard if evita que WriteCache sobrescriba la marca de agua almacenada cuando no se obtuvieron registros esta ejecución.

cacheKey y cacheExpiry deben coincidir con los valores utilizados en la Parte 1. Si el script de actualización se ejecuta en una operación diferente a la del script de lectura, vuelve a asignar estos valores o guárdalos como variables de proyecto para que ambos scripts hagan referencia a la misma clave.

Verifica la integración

1. Establece default_watermark_date a una fecha varios meses en el pasado. Implementa y ejecuta la operación. Agrega llamadas a WriteToOperationLog para registrar el valor de watermarkDate y el número de registros devueltos. Confirma que watermarkDate es igual al valor predeterminado y que la consulta devolvió registros.

2. Ejecuta la operación una segunda vez sin modificar ningún registro de origen. Debido a que el filtro es mayor o igual, el registro (o registros) cuyo LastModifiedDate es igual a la marca de agua almacenada se obtiene nuevamente. Confirma que solo se devuelven esos registros límite (no el conjunto completo), lo que significa que la marca de agua se escribió correctamente después de la primera ejecución, y que el procesamiento idempotente no crea salida duplicada.

3. Modifica un registro de origen y ejecuta la operación nuevamente. Confirma que se devuelve y procesa el registro modificado. Los registros que aún están en el límite de marca de agua anterior también pueden re-obtenerse; el procesamiento idempotente asegura que no produzcan duplicados.

4. Si la marca de agua no persiste entre ejecuciones, confirma que:

   • La clave de caché es idéntica en los scripts de lectura y actualización.
   • Ambos scripts utilizan el mismo alcance ("project").
   • El valor escrito por WriteCache está en el formato ISO 8601 que Salesforce espera en SOQL.

5. Si se obtienen todos los registros en cada ejecución, es posible que la llamada a WriteCache no se esté ejecutando. Confirma que el script de actualización se ejecute después de que se completen todas las operaciones secundarias. Si utilizas RunOperation para encadenar operaciones, coloca el script de actualización en la operación principal después de que regresa la llamada a RunOperation, no dentro de un bucle de transformación.