Saltar al contenido

Procesar registros incrementales utilizando un marcador de posición en Jitterbit Studio

Introducción

Un marcador de posición 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 el marcador almacenado, lo utiliza para filtrar registros ya procesados y actualiza el marcador después de que se completa el lote. Solo se recuperan registros nuevos o modificados en ejecuciones posteriores.

Esta guía demuestra el patrón utilizando casos de Salesforce como fuente. El marcador es la fecha máxima de LastModifiedDate de la ejecución anterior. El patrón se aplica a cualquier fuente que exponga una marca de tiempo de modificación confiable y que soporte consultas filtradas.

Dos funciones de caché gestionan el valor almacenado:

  • ReadCache: Recupera el marcador almacenado al inicio de cada ejecución.
  • WriteCache: Actualiza el marcador después de que se completa el procesamiento.

Para una introducción a ReadCache y WriteCache, incluyendo opciones de alcance de caché y límites de tasa, consulte Detectar y deduplicar registros utilizando funciones hash.

Patrón de diseño

Los pasos de lectura y actualización del marcador 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 hacia adelante automáticamente.

flowchart LR A["Script
Read watermark
from cache"] --> B["Script
Build filtered
SOQL query"] B --> C["Salesforce
Query activity
or SfLookupAll"] C --> D["Process records
(transformation or
child operations)"] D --> E["Script
Update watermark
in cache"]
Paso Propósito
Leer marcador Leer la fecha almacenada; inicializar a un valor predeterminado si no hay nada almacenado.
Construir consulta filtrada Incrustar el marcador en el filtro de consulta.
Consultar fuente Recuperar registros modificados en o después del marcador.
Procesar registros Ejecutar la transformación o cadena de operaciones secundarias.
Actualizar marcador Establecer el marcador a la fecha de modificación máxima entre los registros recuperados y escribirlo en la caché.

Parte 1: Leer el marcador

Agrega un paso de script como el primer paso de la operación. El script lee la marca de agua almacenada y recurre a una fecha predeterminada 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é: Usa una clave que sea única para este conjunto de datos dentro del proyecto. Prefijar con el nombre del proyecto o un identificador del conjunto de datos (por ejemplo, $project_name + "_SF_Case_LM") evita colisiones de claves cuando se almacenan múltiples marcas de agua 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 como para incluir todos los registros que deseas en la primera ejecución. Usa 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 a través de 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 para todas las operaciones en el proyecto y lo persiste entre ejecuciones. Usa "env" si la marca de agua debe compartirse entre múltiples proyectos en el mismo entorno.

Parte 2: Consulta usando la marca de agua

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

Usando SfLookupAll en un script

SfLookupAll devuelve un arreglo bidimensional de registros coincidentes. Úsalo cuando el conjunto de resultados se pase a un script o transformación subsiguiente para su 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.

Usando una actividad de Consulta de Salesforce

Si usas una actividad de Consulta de Salesforce en lugar de SfLookupAll, pasa watermarkDate como una variable de Jitterbit y haz referencia a ella en el campo de condición de la actividad. Establece el operador de condición en mayor o igual que e ingresa [watermarkDate] como el valor.

Utiliza un filtro de mayor o igual y procesamiento idempotente

Un filtro de estrictamente mayor (LastModifiedDate > la marca de agua) puede omitir registros. Cuando varios registros comparten la misma marca de tiempo límite y solo algunos fueron incluidos en el lote anterior, el filtro estricto excluye el resto en la siguiente ejecución, y nunca se procesan. Utiliza un filtro de mayor o igual (>=) para que los registros límite sean re-incluidos. Esto vuelve a recuperar los registros que definieron la marca de agua anterior, por lo que el procesamiento posterior debe ser idempotente (por ejemplo, upsert por una clave única, o deduplicar) para evitar la creación de duplicados. Para un enfoque de deduplicación, consulta Detectar y deduplicar registros utilizando funciones hash.

Parte 3: Actualizar la marca de agua

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

// 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");
);

Deriva la marca de agua de los registros procesados, no del máximo de la fuente

No establezcas la marca de agua volviendo a consultar la fuente por su máximo actual (por ejemplo, SELECT max(LastModifiedDate) FROM Case). Los registros pueden ser modificados en la fuente entre la consulta de la Parte 2 y este paso de actualización. Esos registros no son parte del lote actual, pero una consulta de máximo de fuente incluiría sus marcas de tiempo, avanzando la marca de agua más allá de registros que nunca fueron recuperados. En la siguiente ejecución, el filtro estricto los omite y se pierden permanentemente. Siempre deriva la marca de agua de los registros que esta ejecución realmente recuperó y procesó.

Si utilizas la ruta de actividad Consulta de Salesforce desde Parte 2 en lugar de SfLookupAll, captura el LastModifiedDate máximo a medida que 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 guardia if evita que WriteCache sobrescriba la marca de agua almacenada cuando no se recuperaron registros en 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, asigne estos valores nuevamente o guárdelos como variables de proyecto para que ambos scripts hagan referencia a la misma clave.

Verificar la integración

  1. Establezca default_watermark_date en una fecha de varios meses en el pasado. Despliegue y ejecute la operación. Agregue llamadas a WriteToOperationLog para registrar el valor de watermarkDate y el número de registros devueltos. Confirme que watermarkDate sea igual al valor predeterminado y que la consulta haya devuelto registros.

  2. Ejecute la operación una segunda vez sin modificar ningún registro fuente. Debido a que el filtro es mayor o igual, el registro (o registros) cuyo LastModifiedDate es igual a la marca de agua almacenada se recuperan nuevamente. Confirme que solo se devuelvan 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. Modifique un registro fuente y ejecute la operación nuevamente. Confirme que el registro modificado se devuelva y procese. Los registros que aún están en el límite de la marca de agua anterior también pueden ser recuperados nuevamente; el procesamiento idempotente asegura que no produzcan duplicados.

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

    • La clave de caché es idéntica en ambos scripts, el de lectura y el de 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 recuperan todos los registros en cada ejecución, es posible que la llamada a WriteCache no se esté ejecutando. Confirme que el script de actualización se ejecute después de que todas las operaciones secundarias se completen. Si utiliza RunOperation para encadenar operaciones, coloque el script de actualización en la operación principal después de que la llamada a RunOperation regrese, no dentro de un bucle de transformación.