Inkrementelle Datensätze mit einem High-Watermark in Jitterbit Studio verarbeiten

Einführung

Ein High-Watermark ist ein gespeicherter Wert, der den letzten Punkt kennzeichnet, bis zu dem eine Synchronisierungsoperation Daten verarbeitet hat. Bei jeder Ausführung liest die Operation den gespeicherten Watermark, nutzt ihn zum Filtern bereits verarbeiteter Datensätze und aktualisiert den Watermark nach Abschluss des Batches. Bei nachfolgenden Ausführungen werden nur neue oder geänderte Datensätze abgerufen.

Dieses Handbuch demonstriert das Muster anhand von Salesforce-Cases als Quelle. Der Watermark ist das Maximum von LastModifiedDate aus der vorherigen Ausführung. Das Muster gilt für jede Quelle, die einen zuverlässigen Änderungszeitstempel offenlegt und gefilterte Abfragen unterstützt.

Zwei Cache-Funktionen verwalten den gespeicherten Wert:

• ReadCache: Ruft den gespeicherten Watermark zu Beginn jeder Ausführung ab.
• WriteCache: Aktualisiert den Watermark nach Abschluss der Verarbeitung.

Eine Einführung in ReadCache und WriteCache, einschließlich Cache-Scope-Optionen und Ratenbeschränkungen, finden Sie unter Datensätze mit Hash-Funktionen erkennen und deduplizieren.

Designmuster

Die Schritte zum Lesen und Aktualisieren des Watermark setzen die Hauptlogik für Abfrage und Verarbeitung in Klammern. Das gespeicherte Datum wird nach jeder erfolgreichen Ausführung vorverlegt, sodass sich das Abfragefenster automatisch nach vorne verschiebt.

Teil 1: Den Watermark lesen

Fügen Sie als ersten Schritt der Operation einen Script-Schritt hinzu. Das Script liest den gespeicherten Watermark und greift bei der ersten Ausführung auf ein Standarddatum zurück:

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

Cache-Schlüssel: Verwenden Sie einen Schlüssel, der innerhalb des Projekts eindeutig für diesen Datensatz ist. Die Voranstellung des Projektnamens oder eines Datensatz-Identifiers (z. B. $project_name + "_SF_Case_LM") verhindert Schlüsselkollisionen, wenn mehrere Watermarks im selben Projekt gespeichert werden.

Standarddatum: default_watermark_date ist eine Projektvariable, die in Projektvariablen auf ein vergangenes Datum eingestellt ist, das weit genug zurückliegt, um alle Datensätze einzubeziehen, die Sie bei der ersten Ausführung benötigen. Verwenden Sie das ISO-8601-Format, das Salesforce in SOQL erwartet: z. B. 2000-01-01T00:00:00.000Z.

Ablauf: 30 Tage (2592000 Sekunden) halten den Watermark über geplante Ausführungen hinweg verfügbar. Erhöhen Sie diesen Wert für Operationen, die seltener ausgeführt werden.

Scope: Der "project"-Scope macht den zwischengespeicherten Wert für alle Operationen im Projekt zugänglich und speichert ihn zwischen Ausführungen. Verwenden Sie "env", wenn der Watermark projektübergreifend in der gleichen Umgebung freigegeben werden muss.

Teil 2: Mit dem Watermark abfragen

Nachdem das Script den Watermark gelesen hat, fragen Sie die Quelle mit watermarkDate als Filtergrenze ab.

SfLookupAll in einem Script verwenden

SfLookupAll gibt ein zweidimensionales Array von entsprechenden Datensätzen zurück. Verwenden Sie es, wenn das Ergebnis an ein nachfolgendes Script oder eine Transformation zur Verarbeitung übergeben wird:

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

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

Weitere Informationen zum Erstellen und Ausführen von SOQL-Abfragen finden Sie unter Salesforce-Datensätze mit SOQL abfragen.

Eine Salesforce-Abfrageaktivität verwenden

Wenn Sie stattdessen eine Salesforce-Aktivität Query verwenden, übergeben Sie watermarkDate als Jitterbit-Variable und referenzieren Sie sie im Bedingungsfeld der Aktivität. Stellen Sie den Bedingungsoperator auf größer als oder gleich ein und geben Sie [watermarkDate] als Wert ein.

Teil 3: Wassermarke aktualisieren

Nachdem alle Datensätze im Batch verarbeitet wurden, setzen Sie die Wassermarke auf den maximalen LastModifiedDate-Wert unter den Datensätzen, die tatsächlich in Teil 2 abgerufen wurden, und schreiben Sie ihn in den Cache. Fügen Sie dies als letzten Skriptschritt im Vorgang oder im finalen Vorgang der Kette nach Abschluss aller untergeordneten Vorgänge hinzu:

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

Wenn Sie den Salesforce-Aktivitätspfad Query aus Teil 2 statt SfLookupAll verwenden, erfassen Sie den maximalen LastModifiedDate-Wert während der Verarbeitung der Datensätze (z. B. akkumulieren Sie das Maximum in einer globalen Variablen in der Verarbeitungstransformation), und schreiben Sie diesen Wert hier in den Cache.

Die if-Schutzklausel verhindert, dass WriteCache die gespeicherte Wassermarke überschreibt, wenn in diesem Durchlauf keine Datensätze abgerufen wurden.

cacheKey und cacheExpiry müssen den Werten entsprechen, die in Teil 1 verwendet werden. Falls das Aktualisierungsskript in einem anderen Vorgang als das Leseskript ausgeführt wird, weisen Sie diese Werte neu zu oder speichern Sie sie als Projektvariablen, damit beide Skripte denselben Schlüssel referenzieren.

Integration verifizieren

1. Setzen Sie default_watermark_date auf ein Datum mehrere Monate in der Vergangenheit. Stellen Sie den Vorgang bereit und führen Sie ihn aus. Fügen Sie WriteToOperationLog-Aufrufe hinzu, um den Wert von watermarkDate und die Anzahl der zurückgegebenen Datensätze zu protokollieren. Bestätigen Sie, dass watermarkDate dem Standardwert entspricht und dass die Abfrage Datensätze zurückgegeben hat.

2. Führen Sie den Vorgang ein zweites Mal aus, ohne Quelldatensätze zu ändern. Da der Filter größer-gleich ist, wird der Datensatz (oder die Datensätze), dessen LastModifiedDate der gespeicherten Wassermarke entspricht, erneut abgerufen. Bestätigen Sie, dass nur diese Grenzwertdatensätze zurückgegeben werden (nicht der vollständige Satz), was bedeutet, dass die Wassermarke nach dem ersten Durchlauf korrekt geschrieben wurde, und dass die idempotente Verarbeitung keine Duplikate in der Ausgabe erstellt.

3. Ändern Sie einen Quelldatensatz und führen Sie den Vorgang erneut aus. Bestätigen Sie, dass der geänderte Datensatz zurückgegeben und verarbeitet wird. Datensätze, die sich noch an der vorherigen Wassermarkenbegrenzung befinden, können ebenfalls erneut abgerufen werden. Die idempotente Verarbeitung stellt sicher, dass sie keine Duplikate erzeugen.

4. Wenn die Wassermarke zwischen den Durchläufen nicht beibehalten wird, bestätigen Sie Folgendes:

   • Der Cache-Schlüssel ist in den Lese- und Aktualisierungsskripten identisch.
   • Beide Skripte verwenden denselben Bereich ("project").
   • Der von WriteCache geschriebene Wert hat das ISO-8601-Format, das Salesforce in SOQL erwartet.

5. Wenn alle Datensätze bei jedem Durchlauf abgerufen werden, wird der WriteCache-Aufruf möglicherweise nicht ausgeführt. Bestätigen Sie, dass das Aktualisierungsskript nach Abschluss aller untergeordneten Vorgänge ausgeführt wird. Wenn Sie RunOperation verwenden, um Vorgänge zu verketten, platzieren Sie das Aktualisierungsskript im übergeordneten Vorgang nach der Rückgabe des RunOperation-Aufrufs, nicht innerhalb einer Transformationsschleife.