Processar registros incrementais usando um high-watermark no Jitterbit Studio

Introdução

Um high-watermark é um valor armazenado que marca o ponto mais recente que uma operação de sincronização processou. A cada execução, a operação lê o watermark armazenado, usa-o para filtrar registros já processados e atualiza o watermark após a conclusão do lote. Apenas registros novos ou modificados são recuperados nas execuções subsequentes.

Este guia demonstra o padrão usando casos do Salesforce como fonte. O watermark é o valor máximo de LastModifiedDate da execução anterior. O padrão se aplica a qualquer fonte que exponha um timestamp de modificação confiável e suporte consultas filtradas.

Duas funções de cache gerenciam o valor armazenado:

• ReadCache: Recupera o watermark armazenado no início de cada execução.
• WriteCache: Atualiza o watermark após a conclusão do processamento.

Para uma introdução a ReadCache e WriteCache, incluindo opções de escopo de cache e limites de taxa, consulte Detectar e desduplicar registros usando funções de hash.

Padrão de design

As etapas de leitura e atualização do watermark envolvem a lógica principal de consulta e processamento. A data armazenada avança após cada execução bem-sucedida, então a janela de consulta se desloca automaticamente para frente.

Parte 1: Ler o watermark

Adicione uma etapa de script como a primeira etapa da operação. O script lê o watermark armazenado e retorna a uma data padrão na primeira execução:

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

Chave de cache: Use uma chave exclusiva deste conjunto de dados dentro do projeto. Adicionar como prefixo o nome do projeto ou um identificador de conjunto de dados (por exemplo, $project_name + "_SF_Case_LM") evita colisões de chave quando vários watermarks são armazenados no mesmo projeto.

Data padrão: default_watermark_date é uma variável de projeto definida em Variáveis de projeto para uma data passada suficientemente anterior para incluir todos os registros que você deseja na primeira execução. Use o formato ISO 8601 que o Salesforce espera no SOQL: por exemplo, 2000-01-01T00:00:00.000Z.

Expiração: 30 dias (2592000 segundos) mantém o watermark disponível em execuções agendadas. Aumente este valor para operações que são executadas com menos frequência.

Escopo: O escopo "project" torna o valor em cache acessível a todas as operações no projeto e o persiste entre execuções. Use "env" se o watermark deve ser compartilhado entre vários projetos no mesmo ambiente.

Parte 2: Consultar usando o watermark

Após o script ler o watermark, consulte a fonte usando watermarkDate como limite do filtro.

Usar SfLookupAll em um script

SfLookupAll retorna uma matriz bidimensional de registros correspondentes. Use-a quando o conjunto de resultados for passado para um script ou transformação subsequente para processamento:

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 mais informações sobre como construir e executar consultas SOQL, consulte Consultar registros do Salesforce usando SOQL.

Usar uma atividade Salesforce Query

Se você usar uma atividade Query do Salesforce em vez de SfLookupAll, passe watermarkDate como uma variável do Jitterbit e faça referência a ela no campo de condição da atividade. Defina o operador de condição como maior que ou igual a e insira [watermarkDate] como o valor.

Parte 3: Atualizar a marca d'água

Após processar todos os registros do lote, defina a marca d'água como o máximo de LastModifiedDate entre os registros realmente recuperados na Parte 2 e grave-a em cache. Adicione isso como a última etapa de script na operação, ou na operação final da cadeia após a conclusão de todas as operações filhas:

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

Se usar o caminho da atividade Query do Salesforce da Parte 2 em vez de SfLookupAll, capture o máximo de LastModifiedDate conforme os registros são processados (por exemplo, acumule o máximo em uma variável global na transformação de processamento) e, depois, grave esse valor em cache aqui.

A proteção if impede que WriteCache sobrescreva a marca d'água armazenada quando nenhum registro foi recuperado nesta execução.

cacheKey e cacheExpiry devem corresponder aos valores usados na Parte 1. Se o script de atualização for executado em uma operação diferente do script de leitura, reatribua esses valores ou armazene-os como variáveis de projeto para que ambos os scripts referenciem a mesma chave.

Verificar a integração

1. Defina default_watermark_date como uma data de vários meses atrás. Implante e execute a operação. Adicione chamadas WriteToOperationLog para registrar o valor de watermarkDate e o número de registros retornados. Confirme que watermarkDate é igual ao padrão e que a consulta retornou registros.

2. Execute a operação uma segunda vez sem modificar nenhum registro de origem. Como o filtro é maior-ou-igual, o(s) registro(s) cuja LastModifiedDate é igual à marca d'água armazenada é(são) recuperado(s) novamente. Confirme que apenas esses registros de limite são retornados (não o conjunto completo), o que significa que a marca d'água foi gravada corretamente após a primeira execução, e que o processamento idempotente não cria saída duplicada.

3. Modifique um registro de origem e execute a operação novamente. Confirme que o registro modificado é retornado e processado. Registros ainda no limite da marca d'água anterior também podem ser recuperados novamente; o processamento idempotente garante que não produzam duplicatas.

4. Se a marca d'água não persistir entre execuções, confirme que:

   • A chave de cache é idêntica nos scripts de leitura e atualização.
   • Ambos os scripts usam o mesmo escopo ("project").
   • O valor gravado por WriteCache está no formato ISO 8601 que o Salesforce espera em SOQL.

5. Se todos os registros forem recuperados a cada execução, a chamada WriteCache pode não estar sendo executada. Confirme que o script de atualização é executado após a conclusão de todas as operações filhas. Se usar RunOperation para encadear operações, coloque o script de atualização na operação pai após a chamada RunOperation retornar, não dentro de um loop de transformação.