Skip to main content

UpsertAction

UpsertAction copies data from one input to one output DataObject and keeps the latest version of every record, identified by the primary key of the output table. When a record changes in the source, the version stored so far is overwritten and no history is kept - the pattern known as Slowly Changing Dimension Type 1 (SCD1). In contrast to a plain CopyAction, records which are no longer delivered by the source are kept, so the output always contains the complete set of records ever seen, each one in its most recent state. Use it to build a current state table out of a source you can only read partially, e.g. deltas or incremental loads.

If you need the full history of every change instead of only the latest state, use HistorizeAction, which implements Slowly Changing Dimension Type 2 (SCD2).

Renamed in version 3.0.0

UpsertAction was called DeduplicateAction until version 3.0.0. The old name was misleading: the Action does not deduplicate its input data, it upserts records by primary key. Existing configurations using type = DeduplicateAction keep working, but the name is deprecated - change it to type = UpsertAction, all parameters stay the same.

Requirements​

  • The output DataObject must be a transactional table supporting SQL merge, e.g. DeltaLakeTableDataObject, IcebergTableDataObject, JdbcTableDataObject or SparkConnectTableDataObject.
  • The primary key must be defined on the output table (table.primaryKey). It defines what "the same record" means.
  • The input data must be unique across that primary key. UpsertAction does not deduplicate the input itself, because that is an expensive operation and data is usually unique already. If it is not, add a DeduplicateTransformer to transformers, or you will get errors from the merge statement like DeltaUnsupportedOperationException: [DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE].

Technical columns​

UpsertAction adds one technical column to the output:

ColumnDescription
dl_ts_capturedTimestamp of the last occurrence of the record in the source.

The column name can be changed globally with the SDLB parameter capturedColumnName.

Basic example​

actions {
  upsert-airports {
    type = UpsertAction
    inputId = stg-airports
    outputId = int-airports
  }
}

dataObjects {
  int-airports {
    type = DeltaLakeTableDataObject
    path = "~{id}"
    table {
      db = "default"
      name = "int_airports"
      primaryKey = [ident]
    }
  }
}

Assume the first run at 2024-03-01 04:00:00 gets these two airports from the input stg-airports:

identname
LSZBBern Belp
LSGGGeneva Airport

Both are new, so both are inserted into the output int-airports:

identnamedl_ts_captured
LSZBBern Belp2024-03-01 04:00:00
LSGGGeneva Airport2024-03-01 04:00:00

The second run at 2024-03-02 04:00:00 delivers a new name for LSZB and no longer delivers LSGG:

identname
LSZBBern Belp Airport

The output then contains:

identnamedl_ts_captured
LSZBBern Belp Airport2024-03-02 04:00:00
LSGGGeneva Airport2024-03-01 04:00:00

LSZB was updated to the value and timestamp of the second run, LSGG is still there with the timestamp of the first run, although the source does not deliver it anymore.

Reducing the number of updated records​

By default dl_ts_captured is updated on every execution, even if nothing changed in the source. With a merge statement this rewrites the whole table on every run. Set updateCapturedColumnOnlyWhenChanged = true to update a record only if one of its columns changed:

actions {
  upsert-airports {
    type = UpsertAction
    inputId = stg-airports
    outputId = int-airports
    updateCapturedColumnOnlyWhenChanged = true
  }
}

dl_ts_captured then holds the timestamp of the last change of the record instead of the last time it was seen.

Reading the existing data for the merge can be limited further with mergeModeAdditionalJoinPredicate, e.g. if it is sufficient to consider records captured within the last 7 days. Use the table alias existing to reference columns of the existing table data:

    mergeModeAdditionalJoinPredicate = "existing.dl_ts_captured > current_date - interval 7 days"

Following the time axis of the source system​

By default dl_ts_captured is set to the reference timestamp of the run, so it reflects the schedule of the pipeline rather than the source system. The reference timestamp defaults to the start time of the run and can be overridden with the SDLB parameter referenceTimestamp, e.g. referenceTimestamp = "2024-01-01 00:00:00". If the input contains the timestamp of the last change of a record, e.g. last_updated, set sourceTimestampColumn to use that value instead:

actions {
  upsert-airports {
    type = UpsertAction
    inputId = stg-airports
    outputId = int-airports
    sourceTimestampColumn = last_updated
  }
}
  • The column must exist in the data to upsert and be of type timestamp. Records where it is null fall back to the reference timestamp of the run.
  • There is no auto detection: the column must be configured explicitly.
  • The column itself is not written to the output DataObject, as its value is kept in dl_ts_captured. Copy it to a column with another name in a transformer if you want to keep it.
  • Records arriving late, e.g. having a source timestamp older than the record already stored, are not applied, so that dl_ts_captured always holds the latest version according to the source system.
  • updateCapturedColumnOnlyWhenChanged is normally not needed then, as dl_ts_captured is moved forward only if the source system changed the record anyway. What it changes is that existing records are updated only if the source timestamp increased, instead of comparing all columns. This avoids rewriting records which the source system delivers again with an unchanged timestamp, but a change which the source system did not timestamp is not applied.

Transformations and execution modes​

transformers are applied before the upsert, so they see the input data without dl_ts_captured, see Transformations. Any ExecutionMode can be combined with UpsertAction; DataObjectStateIncrementalMode is a natural fit, as only new or changed records need to be read from the source.

Parameters​

The complete list of parameters is available in the Configuration Schema Viewer. The most relevant ones:

ParameterDefaultDescription
inputId-Input DataObject.
outputId-Output DataObject, a transactional table with defined primary key supporting merge.
transformers[]Transformations to apply before the upsert, in the order of the list.
updateCapturedColumnOnlyWhenChangedfalseUpdate dl_ts_captured only if the record changed in the source, instead of on every execution.
sourceTimestampColumn-Column holding the timestamp of the last change in the source system, used as value for dl_ts_captured.
mergeModeAdditionalJoinPredicate-Condition to limit the existing data read for the merge, using table alias existing.
ignoreOldDeletedColumnsfalseRemove no longer existing columns on schema evolution.
ignoreOldDeletedNestedColumnstrueRemove no longer existing columns from nested data types on schema evolution.

See also​