Verarbeiten inkrementeller Datensätze mit einem Hochwasserzeichen in Jitterbit Studio

Einführung

Ein Hochwasserzeichen ist ein gespeicherter Wert, der den aktuellsten Punkt markiert, den eine Synchronisationsoperation verarbeitet hat. Bei jedem Durchlauf liest die Operation das gespeicherte Wasserzeichen, verwendet es, um bereits verarbeitete Datensätze herauszufiltern, und aktualisiert das Wasserzeichen, nachdem der Batch abgeschlossen ist. Nur neue oder geänderte Datensätze werden bei nachfolgenden Durchläufen abgerufen.

Dieser Leitfaden demonstriert das Muster anhand von Salesforce-Fällen als Quelle. Das Wasserzeichen ist das maximale LastModifiedDate vom vorherigen Durchlauf. Das Muster gilt für jede Quelle, die einen zuverlässigen Änderungszeitstempel bereitstellt und gefilterte Abfragen unterstützt.

Zwei Cache-Funktionen verwalten den gespeicherten Wert:

• ReadCache: Ruft das gespeicherte Wasserzeichen zu Beginn jedes Durchlaufs ab.
• WriteCache: Aktualisiert das Wasserzeichen, nachdem die Verarbeitung abgeschlossen ist.

Für eine Einführung in ReadCache und WriteCache, einschließlich Cache-Bereichsoptionen und Ratenlimits, siehe Datensätze erkennen und deduplizieren mit Hash-Funktionen.

Entwurfsmuster

Die Schritte zum Lesen und Aktualisieren des Wasserzeichens umrahmen die Hauptlogik für Abfrage und Verarbeitung. Das gespeicherte Datum wird nach jedem erfolgreichen Durchlauf vorangetrieben, sodass sich das Abfragefenster automatisch verschiebt.

Teil 1: Wasserzeichen lesen

Fügen Sie einen Skript-Schritt als ersten Schritt der Operation hinzu. Das Skript liest den gespeicherten Wasserzeichen und greift beim ersten Durchlauf 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 für diesen Datensatz innerhalb des Projekts einzigartig ist. Das Voranstellen des Projektnamens oder einer Datensatzkennung (zum Beispiel, $project_name + "_SF_Case_LM") verhindert Schlüsselkonflikte, wenn mehrere Wasserzeichen im selben Projekt gespeichert sind.

Standarddatum: default_watermark_date ist eine Projektvariable die in Projektvariablen auf ein vergangenes Datum gesetzt wird, das weit genug zurückliegt, um alle Datensätze zu umfassen, die Sie beim ersten Durchlauf möchten. Verwenden Sie das ISO 8601-Format, das Salesforce in SOQL erwartet: zum Beispiel, 2000-01-01T00:00:00.000Z.

Ablauf: 30 Tage (2592000 Sekunden) hält das Wasserzeichen über geplante Durchläufe hinweg verfügbar. Erhöhen Sie diesen Wert für Operationen, die weniger häufig ausgeführt werden.

Bereich: Der Bereich "project" macht den zwischengespeicherten Wert für alle Operationen im Projekt zugänglich und bewahrt ihn zwischen den Durchläufen. Verwenden Sie "env", wenn das Wasserzeichen über mehrere Projekte in derselben Umgebung geteilt werden muss.

Teil 2: Abfrage mit dem Wasserzeichen

Nachdem das Skript das Wasserzeichen gelesen hat, fragen Sie die Quelle unter Verwendung von watermarkDate als Filtergrenze ab.

Verwendung von SfLookupAll in einem Skript

SfLookupAll gibt ein zweidimensionales Array von übereinstimmenden Datensätzen zurück. Verwenden Sie es, wenn die Ergebnismenge an ein nachfolgendes Skript 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);

Für weitere Informationen zum Erstellen und Ausführen von SOQL-Abfragen siehe Salesforce-Datensätze mit SOQL abfragen.

Verwendung einer Salesforce-Abfrageaktivität

Wenn Sie eine Salesforce Abfrage-Aktivität anstelle von SfLookupAll verwenden, übergeben Sie watermarkDate als Jitterbit-Variable und verweisen Sie darauf im Bedingungsfeld der Aktivität. Setzen Sie den Bedingungsoperator auf größer als oder gleich und geben Sie [watermarkDate] als Wert ein.

Teil 3: Aktualisieren Sie das Wasserzeichen

Nachdem alle Datensätze im Batch verarbeitet wurden, setzen Sie das Wasserzeichen auf das maximale LastModifiedDate unter den tatsächlich in Teil 2 abgerufenen Datensätzen und schreiben Sie es in den Cache. Fügen Sie dies als letzten Schritt im Skript der Operation hinzu oder in der finalen Operation der Kette, nachdem alle untergeordneten Operationen abgeschlossen sind:

// 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 Query-Aktivitätsweg aus Teil 2 anstelle von SfLookupAll verwenden, erfassen Sie das maximale LastModifiedDate, während die Datensätze verarbeitet werden (zum Beispiel, indem Sie das Maximum in einer globalen Variablen in der Verarbeitungstransformation akkumulieren), und schreiben Sie diesen Wert hier in den Cache.

Der if-Wächter verhindert, dass WriteCache den gespeicherten Wasserstand überschreibt, wenn in diesem Durchlauf keine Datensätze abgerufen wurden.

cacheKey und cacheExpiry müssen mit den Werten übereinstimmen, die in Teil 1 verwendet wurden. Wenn das Aktualisierungsskript in einer anderen Operation als das Leseskript ausgeführt wird, weisen Sie diese Werte entweder erneut zu oder speichern Sie sie als Projektvariablen, damit beide Skripte denselben Schlüssel referenzieren.

Überprüfen Sie die Integration

1. Setzen Sie default_watermark_date auf ein Datum, das mehrere Monate in der Vergangenheit liegt. Bereitstellen und ausführen Sie die Operation. 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 die Operation ein zweites Mal aus, ohne irgendwelche Quelldatensätze zu ändern. Da der Filter größer-gleich ist, wird der Datensatz (oder die Datensätze), dessen LastModifiedDate dem gespeicherten Wasserstand entspricht, erneut abgerufen. Bestätigen Sie, dass nur diese Grenzdaten zurückgegeben werden (nicht die vollständige Menge), was bedeutet, dass der Wasserstand nach dem ersten Durchlauf korrekt geschrieben wurde und dass idempotente Verarbeitung keine doppelten Ausgaben erzeugt.

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

4. Wenn der Wasserstand zwischen den Durchläufen nicht erhalten bleibt, bestätigen Sie, dass:

   • Der Cache-Schlüssel in beiden Skripten (Lesen und Aktualisieren) identisch ist.
   • Beide Skripte denselben Geltungsbereich ("project") verwenden.
   • Der von WriteCache geschriebene Wert im ISO 8601-Format vorliegt, das Salesforce in SOQL erwartet.

5. Wenn bei jedem Durchlauf alle Datensätze abgerufen werden, wird der WriteCache-Aufruf möglicherweise nicht ausgeführt. Bestätigen Sie, dass das Aktualisierungsskript nach Abschluss aller untergeordneten Operationen ausgeführt wird. Wenn Sie RunOperation verwenden, um Operationen zu verketten, platzieren Sie das Aktualisierungsskript in der übergeordneten Operation, nachdem der RunOperation-Aufruf zurückgegeben wurde, nicht innerhalb einer Transformationsschleife.