Deduplicate Data Exported From Reactor in Your Data Warehouse

Contents

1. Overview
2. Example Deduplication Queries
   1. Snowflake
   2. Google BigQuery

Overview

After collecting, indexing, transforming, and outputting data from your source system(s) to your data warehouse, Reactor inserts the new data into your data warehouse as new rows of data. This differs from other systems such as Fivetran, which sync or update existing records in the data warehouse. Over time, multiple rows representing different updates to the same record can accumulate in the data warehouse.

To avoid data duplication and ensure that only the latest version of each record is used for analytical purposes, we recommend creating a deduplication view on top of the landing table. This view will filter out outdated rows and retain only the most recent version of each record for accurate and efficient data analysis.

Below are example deduplication queries that can be used in Google BigQuery and Snowflake, respectively.

Example Deduplication Queries

Google BigQuery

with deduped_records as (
  select  *,  -- The following two fields are used to identify the most recent version of a record.
  -- The message_version field is used to identify the most recent version of an identity and its versions.
  -- Using this field, we filter out Reactor Replays, network crashes, checkpointing, and mapping changes.
  row_number() over (
  partition by
  node, -- Origin of the source data
  scope, -- Label representing the brand, store, or business unit
  source_id, -- An entity field of the source record.
  -- You'll want to modify this to the entity field of your landing table.
  -- Example: order_id, product_id, user_id, etc.
  source_message_version -- A version field of the source record (if available)
  -- If your landed data set has a timestamp-like field, that can be used to determine the version of the record.
  -- e.g., if you are tracking a Type 2 SCD history for an entity
  order by version desc, parsed_timestamp desc, loaded_at desc
  ) as message_version,  -- The entity_version field is used to identify the most recent version of a record entity.
  -- You should not dedupe on this field as a default, but you can use it as a filter in a downstream model.
  -- Example: `select * from vw_landed_dedupe where entity_version = 1`
  row_number() over (
  partition by
  node, -- Origin of the source data
  scope, -- Label representing the brand, store, or business unit
  regexp_substr(scid, '[^:]*$') -- entity id
  order by  source_version desc, -- if available
  version desc, parsed_timestamp desc, loaded_at desc
  ) as entity_version
  from `BigQueryProject.Dataset.landing_table` -- Replace this with the BigQueryProject/Database/landing table in your environment
) 
select
  * except(message_version) 
from deduped_records 
where message_version = 1

Snowflake

with deduped_records as (
  select  *,  -- The following two fields are used to identify the most recent version of a record.
  -- The message_version field is used to identify the most recent version of an identity and its versions.
  -- Using this field, we filter out Reactor Replays, network crashes, checkpointing, and mapping changes.
  row_number() over (
  partition by
  node, -- Origin of the source data
  scope, -- Label representing the brand, store, or business unit
  source_id, -- An entity field of the source record
  -- You'll want to modify this to the entity field of your landing table
  -- Example: order_id, product_id, user_id, etc.
  source_message_version -- A version field of the source record (if available)
  -- If your landed data set has a timestamp-like field, that can be used to determine the version of the record.
  -- e.g., if you are tracking a Type 2 SCD history for an entity.
   order by version desc, parsed_timestamp desc, loaded_at desc
  ) as message_version,  -- The entity_version field is used to identify the most recent version of a record entity.
  -- You should not dedupe on this field as a default, but you can use it as a filter in a downstream model.
  -- Example: `select * from vw_landed_dedupe where entity_version = 1`
  row_number() over (
  partition by
  node, -- Origin of the source data
  scope, -- Label representing the brand, store, or business unit
  regexp_substr(scid, '[^:]*$') -- entity id
  order by  source_version desc, -- if available
  version desc, parsed_timestamp desc, loaded_at desc
  ) as entity_version
  from database.schema.landing_table -- Replace this with the database/schema/landing table in your environment
) 
select
  * exclude(message_version) 
from deduped_records 
where message_version = 1