Change Replication using High Watermark (HWM)

About High Watermark Change Replication (HWM)

In the Etlworks Integrator, it is possible to create a Flow that tracks changes in a data source (for example, database or API) and loads the changed records into the destination (for example, another database or API). Another name for change replication is data synchronization.

The Etlworks Integrator supports unidirectional change replication: from the source to the destination. To replicate changes in the opposite direction (from the destination to the source), a separate Flow must be created, where the source and the destination swap places.

This technique uses a High Watermark Field to identify recently changed records. 

Read about other change replication techniques available in the Etlworks Integrator.

What you need to know

Use Cases

Change replication can occur:

• Between two databases: a Flow pulls changed records from the source database to update the destination database.

• Between database and any destination: a Flow pulls changed records from the source database and loads changes into any destination, for example, creates update files in Amazon S3.

• Between any source and any destination: a Flow pulls changed records from any source, for example, web service, and loads them into any supported destination, for example, data warehouse.

Resolve change conflicts

When the same record is modified in both the source and destination, the Etlworks Integrator will simply apply the source changes to the destination.

High Watermark Field

A high watermark is the highest peak in value that a field has reached.

Which fields could become a High Watermark Field?

Basically, any field can be a High Watermark Field as long as it has the data type TIMESTAMP, DATE, or NUMERIC and can be used to uniquely identify recent changes.

For a database, it is recommended that there is an index for the High Watermark Field.

Is it possible to set two or more fields as a high watermark for a single transformation?

There can only be one High Watermark Field.

Example of the High Watermark Field

Suppose there is a table audit_trail in a source PostgreSQL database. This table will be updated each time you log into the system or executes a specific function. We would like to track changes in this table and load them into an online data warehouse (which is also a database).

Suppose that the table audit_trail has a field Last_modified TIMESTAMP NOT NULL which is updated each time a record is inserted or updated. We would use Last_modified as the High Watermark Field so we can track changes in the audit_trail table. 

Use a fully qualified field name for the High Watermark Field

If a database is a source for the change replication, it is possible to use a fully qualified field name as the value for the High Watermark Field, as in the example below:

public.audit_trail.partition_date

It makes sense to enter a fully qualified field name if the Source query, which drives the change replication, includes JOINs, sub-selects, and table aliases so that field name collisions can be avoided.

How HWM change replication works

1. The Etlworks Integrator always tracks what records have been extracted and what records have been loaded during a typical ETL.

2. It is possible to set up a High Watermark Field for that particular transformation.

3. When a High Watermark Field is set, the Etlworks Integrator calculates a maximum value for the field (high watermark) for the last successful load.

4. The Etlworks Integrator stores this value with the metadata.

5. The next time the ETL process is executed, the Etlworks Integrator uses the previously calculated high watermark value to filter out records that are older than the previous high watermark.

6. This technique works equally well for databases, files, and APIs.

7. If the change replication is from a database, the Etlworks Integrator modifies the WHERE clause on the fly, to select records that are newer than the high watermark select * from audit_trail where last_modified > ? where ? is a high watermark value.

8. If the change replication is from an API or file, the Etlworks Integrator automatically adds a JavaScript filter that skips all records that are older than the high watermark.

Calculate high watermark field value

Calculate high watermark field value using SQL

This technique works only if the destination is a database.

The Etlworks Integrator automatically tracks the changes in the source and replicates them in the destination. While doing this, it automatically calculates the highest peak in value of the High Watermark Field and uses it to pull only the changed records from the source. It all works perfectly as long the destination table is updated only by running that specific change replication Flow. Unfortunately, that is not always the case.

To make a change replication Flow bulletproof, you can specify an SQL query that will be calculating the High Watermark Field Value on the fly, based on the current state of the destination table. Basically, it will be pulling only the records in the source which do not exist in the destination or have changed in the source since the last run of the Flow. 

HWM query

Just enter an SQL query that returns the current maximum field value in the High Watermark Field Value:

In this example, the High Watermark Field Value is calculated using the SQL query below:

SELECT max(audit_trail_id) FROM audit_trail_updates

Instead of the actual table name of the FROM clause, you can use the token {TABLE}. It is specifically useful when extracting and loading data from multiple database objects using a wildcard source name.

SELECT max(id) FROM {TABLE}

The query is executed on the destination Connection.

Ignore exceptions when executing HWM query

The default behavior when executing an HWM query is to throw an exception if the quire is executed with an error.

There is an edge case when the system is trying to execute an HWM query against a table that does not exist yet. For example, if you configured a transformation to automatically create a destination table and the table doesn't exist yet at the time when the HWM query is executed.

To handle the edge case, use the field High Watermark Exception to Ignore. You can enter a string, for example, object doesn't exist, which, if it is found in the exception, will set the HWM value to null instead of throwing an exception.

Calculate high watermark field value using JavaScript

Using this technique, you can change the HWM value on the fly by providing a JavaScript program that returns a new HWM value. The technique is typically used together with Change Data Tracking.

You can enter the program in the Change High Watermark Field Value field located under MAPPING > Change Replication tab.

Example

highWatermark == null ? 0 : java.lang.Math.round(highWatermark)

The code above sets the initial HWM value to 0 and then returns the long value of the highWatermark (internally, it is stored as a decimal number).

The last evaluated line in the code returns the new HWM value.

Available objects

The following objects can be referenced by name from JavaScript code:

Use previous high watermark value in source query and high watermark query

The High Watermark Value for the previous run is always stored in the metadata storage. It can be used in the Source query and query to calculate the high watermark value (HWM query).

Reference HWM value in the query

Use the following notation to reference the HWM value in the query:

{TRANSFORMATION_NAME_HIGH_WATERMARK}

Read how to set the transformation name.

Work with NULL HWM value

To protect the query against the NULL HWM value, use the following trick:

{token} is null or field > {token}

If the value of the HWM field represented by {token} is indeed NULL, the query will be automatically transformed to:

NULL is null or field > NULL

Use the previous HWM value as the Source query

You can add the following string anywhere in the Source query:

{TRANSFORMATION_NAME_HIGH_WATERMARK} is null OR 
hwm_field > {TRANSFORMATION_NAME_HIGH_WATERMARK}

For example, when:

• The Source query is select * from audit
• The Transformation Nameis AUDIT
• The HWM field is audit_id

Then, the Source query will be:

select * from audit 
WHERE {AUDIT_HIGH_WATERMARK} is null OR audit_id > {AUDIT_HIGH_WATERMARK}

Use the previous HWM value in the HWM query

Anywhere in the HWM query, you can add the following string:

{TRANSFORMATION_NAME_HIGH_WATERMARK} is null OR 
hwm_field > {TRANSFORMATION_NAME_HIGH_WATERMARK}

For example, when:

• The HWM query is select max(audit_id) from audit_updates
• The Transformation Name is AUDIT
• The HWM field is audit_id

Then, the HWM query will be:

select max(audit_id) from audit_updates 
WHERE {AUDIT_HIGH_WATERMARK} is null OR audit_id > {AUDIT_HIGH_WATERMARK}

Manually set the value of the high watermark field

If the destination table is not empty, or it is possible that the records in the destination are getting updated by the different processes (in addition to the change replication Flow), it makes sense to manually set the value of the High Watermark Field.

There are two fields that can be used for that purpose:

• High Watermark Field Value on First Run: the optional value of the High Watermark Field when running change replication scenario for the first time. If the High Watermark Field is TIMESTAMP, use the YYYY-MM-DD HH:mm:SS.mil Format, for example, 2018-08-01 19:38:31.377. The value set for this field will be used only when the change replication Flow is running the first time and will be ignored on the next run.
• High Watermark Field Value : the optional value of the High Watermark Field when running change replication scenario. If the High Watermark Field is TIMESTAMP, use the YYYY-MM-DD HH:mm:SS.mil Format, for example, 2018-08-01 19:38:31.377. This value overrides the value set automatically when running the scenario and the value used for the first run. You can also enter an SQL query that returns the current maximum value of the High Watermark Field in the destination table. For example, select max(last_updated) from order. Read more about calculating high watermark field value.

HWM replication when the source is configured with a wildcard

The Etlworks Integrator supports HWM replication when the source is configured with a wildcard. First, the Flow lists all the objects which names match the wildcard, it then creates a separate source-to-destination transformation for each matching source object: database table file, etc. It applies the same HWM rules to each source-to-destination pair. Here's how you can do this:

Step 1. Add a source-to-destination transformation when the source is configured with a wildcard.

Step 2. Click MAPPING and select the Parameters tab. Enter {source} in the Transformation Name field. The token {source} enables a mechanism that sets the Transformation Name to the fully qualified source table name.  Without it, the Transformation Name will be generated dynamically, which will have a negative impact on how the Flow locates the stored HWM value. The other available token is {destination}. You can use both tokens together: {source} to {destination}. We recommend {source}.

Step 3. Set the value of the High Watermark Field under the Change Replication tab. 

It is expected that all source tables have the same High Watermark Field, for example, last_update.

Step 4. Optionally, set the SQL to calculate the High Watermark Field value on each run. Use token {table} for the destination table name.

Incremental synchronization between any source and any destination

If your source is not a database, you can still set up a change replication Flow which will load only the records that have changed since the last successful load.

Basically, all you need to do is to set the High Watermark Field and enable the change replication for the transformation.

Make sure that the High Watermark Field is either TIMESTAMP / DATE (or can be converted to TIMESTAMP / DATE), or NUMERIC.