# Deduplicating Company Records in a Multi-Source Data Centralization Project — Rittman Analytics URL: https://rittmananalytics.com/blog/2021/01/17/deduplicating-company-records-in-a-multi-source-data-centralization-project-rittman-analytics Description: Deduplicating Company Records in a Multi-Source Data Centralization Project using dbt, Google BigQuery or Snowflake Modern Data Stack 17 Jan Written By Mark Rittman One of the most common tasks… Author: Mark Rittman Published: 2021-01-17T01:00:00+01:00 Modified: 2021-01-17T01:00:00+01:00 --- ![Deduplicating Company Records in a Multi-Source Data Centralization Project — Rittman Analytics](https://rittmananalytics.com/api/public/og-image?src=https%3A%2F%2Fi0.wp.com%2Fdtdtemp8.wpcomstaging.com%2Fwp-content%2Fuploads%2F2021%2F01%2F1752872505888-vztl9t.png%3Ffit%3D1400%252C781%26ssl%3D1) # Deduplicating Company Records in a Multi-Source Data Centralization Project using dbt, Google BigQuery or Snowflake [Modern Data Stack](https://rittmananalytics.com/blog/category/Modern+Data+Stack) 17 Jan Written By [Mark Rittman](https://rittmananalytics.com/blog?author=63c478ac6811232b3153d585) One of the most common tasks in a [data centralization project](https://rittmananalytics.com/data-centralization) is to create single, deduplicated records for each of the companies, contacts, products and other entities the business interacts with. Doing this allows you to connect sales activity from Salesforce and Hubspot to project delivery and invoicing data from Jira and Xero, for example: ![](https://rittmananalytics.com/api/public/og-image?src=https%3A%2F%2Fi0.wp.com%2Fnfqvbiqopvwmhnrrabjr.supabase.co%2Fstorage%2Fv1%2Fobject%2Fpublic%2Fblog-images%2Fimported-images%2F1752872505888-vztl9t.png%3Fresize%3D1400%252C781%26ssl%3D1) cross-company.png Doing this in-practice can however get pretty confusing and complicated quickly as a [recent thread on the dbt Slack forum discussed](https://getdbt.slack.com/archives/C0VLZPLAE/p1610712146023700). To summarise the complications this adds to a typical data centralization project: - Assuming you’re going to create a single record for each company, contact, product and so on, how do you store the source system IDs for those entities in such a way that you can link the incoming deals, orders, invoices and payments from those source systems to these new records? - What happens if not all of the sources of company data, for example, provide all of the fields that the main source provides? A CRM source such as HubSpot will typically provide dozens of data fields for each company whereas others may only provide a name and billing address - What if one source has the concept of a company or organisation in how it works but doesn’t provide a table of companies in its data extract; for example when you’re working with Jira as a data source and each project is named after one of your clients, but there’s only project and issue tables in the data export and no explicit “company” table - And what if two of the companies, contacts or products you need to deduplicate don’t naturally merge together on the name, code or other field you use for this purpose? All of this is solvable with enough thinking, effort and coding but doing so whilst keeping your project agile, delivering fast and not becoming a spaghetti mess of hacked-together code can be a challenge. Here’s how we do this on our own client [data centralization projects](https://rittmananalytics.com/data-centralization), using Google BigQuery or Snowflake as the data warehousing platform, [dbt (“Data Build Tool”)](https://rittmananalytics.com/getting-started-with-dbt) and our [open-source RA Warehouse for dbt](https://rittmananalytics.com/ra-warehouse-for-dbt) framework [available on Github](https://github.com/rittmananalytics/ra_data_warehouse) as a public repo. ## Step 1 : Standardise Entity Staging Data Sources Customers, contacts, projects and other shared dimensions are automatically created from all data sources, deduplicating by name and merge lookup files using a process that preserves source system keys whilst assigning a unique ID for each customer, contact etc, as shown in the diagram below. ![](https://rittmananalytics.com/api/public/og-image?src=https%3A%2F%2Fi0.wp.com%2Fnfqvbiqopvwmhnrrabjr.supabase.co%2Fstorage%2Fv1%2Fobject%2Fpublic%2Fblog-images%2Fimported-images%2F1752872505917-d3u8nw.png%3Fresize%3D1400%252C462%26ssl%3D1) method.png Each dbt staging module (a folder within a dbt project) provides a unique ID, prefixed with the source name, and another field value (for example, user name) that can be used for deduplicating dimension members downstream. WITH source AS ({{ filter\_stitch\_relation(relation=var(‘stg\_hubspot\_crm\_stitch\_companies\_table’),unique\_column=’companyid’) }}),renamed AS (SELECTCONCAT(‘{{ var(‘stg\_hubspot\_crm\_id-prefix’) }}’,companyid) AS company\_id,REPLACE(REPLACE(REPLACE(properties.name.value, ‘Limited’, ”), ‘ltd’, ”),’, Inc.’,”) AS company\_name,properties.address.value AS company\_address,properties.address2.value AS company\_address2,properties.city.value AS company\_city,properties.state.value AS company\_state,properties.country.value AS company\_country,properties.zip.value AS company\_zip,properties.phone.value AS company\_phone,properties.website.value AS company\_website,properties.industry.value AS company\_industry,properties.linkedin\_company\_page.value AS company\_linkedin\_company\_page,properties.linkedinbio.value AS company\_linkedin\_bio,properties.twitterhandle.value AS company\_twitterhandle,properties.description.value AS company\_description,CAST (NULL AS STRING) AS company\_finance\_status,cast (null as string) as company\_currency\_code,properties.createdate.value AS company\_created\_date,properties.hs\_lastmodifieddate.value company\_last\_modified\_dateFROMsource)SELECT\*FROMrenamed As long as a data source can provide an ID and a field by which you can connect its records to the ones from the other source systems, you can just pass null values for the ones that are missing, like this: WITH source AS ({{ filter\_stitch\_relation(relation=var(‘stg\_jira\_projects\_stitch\_projects\_table’),unique\_column=’id’) }}),renamed as (select \* from (SELECTconcat(‘{{ var(‘stg\_jira\_projects\_id-prefix’) }}’,replace(name,’ ‘,’\_’)) AS company\_id,name AS company\_name,cast (null as string) as company\_address,cast (null as string) AS company\_address2,cast (null as string) AS company\_city,cast (null as string) AS company\_state,cast (null as string) AS company\_country,cast (null as string) AS company\_zip,cast (null as string) AS company\_phone,cast (null as string) AS company\_website,cast (null as string) AS company\_industry,cast (null as string) AS company\_linkedin\_company\_page,cast (null as string) AS company\_linkedin\_bio,cast (null as string) AS company\_twitterhandle,cast (null as string) AS company\_description,cast (null as string) as company\_finance\_status,cast (null as string) as company\_currency\_code,cast (null as timestamp) as company\_created\_date,cast (null as timestamp) as company\_last\_modified\_dateFROM source ){{ dbt\_utils.group\_by(n=19) }})select \* from renamed If you’re wondering what [{{ filter\_stitch\_relation … }}](https://github.com/rittmananalytics/ra_data_warehouse/blob/master/macros/filter_stitch_relation.sql) is referring to, it’s a [dbt macro](https://docs.getdbt.com/docs/building-a-dbt-project/jinja-macros/) we use to filter out the historic (append-only) version of rows you get when using Stitch to replicate SaaS data sources into Google BigQuery: {%- macro filter\_stitch\_relation(relation, unique\_column) -%}SELECT\*FROM(SELECT\*,MAX(\_sdc\_batched\_at) OVER (PARTITION BY {{ unique\_column }} ORDER BY \_sdc\_batched\_at RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS max\_sdc\_batched\_atFROM{{ relation }})WHERE\_sdc\_batched\_at = max\_sdc\_batched\_at{%- endmacro -%} ## Step 2 : Union Staging Data Sources Together A variable we’ve created within the [dbt\_project.yml](https://github.com/rittmananalytics/ra_data_warehouse/blob/master/dbt_project.yml.bigquery_example) lists of all the staging data sources that provide company records, named in a standard way so that if we prepend “stg” and append “\_staging” to the start of the data source name it equates to the name of the module (folder) within the dbt project. crm\_warehouse\_company\_sources: \[‘hubspot\_crm’,’harvest\_projects’,’xero\_accounting’,’stripe\_payments’,’asana\_projects’,’jira\_projects’ The [“Integration”](https://github.com/rittmananalytics/ra_data_warehouse/blob/master/models/integration/int_companies_pre_merged.sql) module for company data sources then uses this list of sources provided by the variable to drive a Jinja “for” loop that unions these sources together into a single CTE (common table expression) with t\_companies\_pre\_merged as ({% for source in var(‘crm\_warehouse\_company\_sources’) %}{% set relation\_source = ‘stg\_’ + source + ‘\_companies’ %}select'{{source}}’ as source,\*from {{ ref(relation\_source) }}{% if not loop.last %}union all{% endif %}{% endfor %}) ## Step 3 : Group and Deduplicate on a Common Field This unioned-but-not-yet-deduplicated set of all company records from all sources is then grouped (deduplicated) on a field that contains the same value for the company across all the different sources, such as the company name. grouped as (SELECTcompany\_name,max(company\_phone) as company\_phone,max(company\_website) as company\_website,max(company\_industry) as company\_industry,max(company\_linkedin\_company\_page) as company\_linkedin\_company\_page,max(company\_linkedin\_bio) as company\_linkedin\_bio,max(company\_twitterhandle) as company\_twitterhandle,max(company\_description) as company\_description,max(company\_finance\_status) as company\_finance\_status,max(company\_currency\_code) as company\_currency\_code,min(company\_created\_date) as company\_created\_date,max(company\_last\_modified\_date) as company\_last\_modified\_datefrom t\_companies\_pre\_mergedgroup by 1 ## Step 4 : Store IDs and Multi-Value Fields in Arrays For fields where we’d need to record a number of values for each company, for example when recording the various source system IDs we’ll need to have available when linking invoices coming from Xero, deals from Hubspot and payments from Stripe, we make use of [repeating columns](https://cloud.google.com/bigquery/docs/nested-repeated) when working with Google BigQuery target data warehouses, like this: all\_company\_ids as (SELECT company\_name, array\_agg(distinct company\_id ignore nulls) as all\_company\_idsFROM t\_companies\_pre\_mergedgroup by 1), and repeating nested columns (aka structs) when storing multi-field arrays of related values, for example when storing the components of an address: all\_company\_addresses as (SELECT company\_name, array\_agg(struct(company\_address,company\_address2,company\_city,company\_state,company\_country,company\_zip)ignore nulls) as all\_company\_addresses If the target warehouse is Snowflake then we’d achieve the same result by using a VARIANT data type to store the repeating fields as JSON. all\_company\_ids as (SELECT company\_name,array\_agg(distinct company\_id) as all\_company\_idsFROM t\_companies\_pre\_mergedgroup by 1),all\_company\_addresses as (SELECT company\_name,array\_agg(parse\_json (concat(‘{“company\_address”:”‘,company\_address,'”, “company\_address2″:”‘,company\_address2,'”, “company\_city”:”‘,company\_city,'”, “company\_state”:”‘,company\_state,'”, “company\_country”:”‘,company\_country,'”, “company\_zip”:”‘,company\_zip,'”} ‘))) as all\_company\_addresses ## Step 5: Use a Lookup File for Manual Entity Merging For dimensions where merging of members by name is not sufficient (for example, company names that cannot be relied on to always be spelt the same across all sources) we add a [seed file](https://github.com/rittmananalytics/ra_data_warehouse/blob/master/data/companies_merge_list.csv) to provide manual lookups to map one member to another via their source system IDs, for example: company\_id,old\_company\_idhubspot-3423423,xero-123121hubspot-2412121,stripe-214122xero-123121,salesforce-12312412 and then [extend the logic of the previous grouping step](https://github.com/rittmananalytics/ra_data_warehouse/blob/master/models/integration/int_companies.sql) to make use of this merge file, for example when BigQuery is the target warehouse: from companies\_pre\_merged cleft outer join (select company\_name,ARRAY(SELECT DISTINCT xFROM UNNEST(all\_company\_ids) AS x) as all\_company\_idsfrom (select company\_name, array\_concat\_agg(all\_company\_ids) as all\_company\_idsfrom (select \* from (selectc2.company\_name as company\_name,c2.all\_company\_ids as all\_company\_idsfrom {{ ref(‘companies\_merge\_list’) }} mjoin companies\_pre\_merged c1 on m.old\_company\_id in UNNEST(c1.all\_company\_ids)join companies\_pre\_merged c2 on m.company\_id in UNNEST(c2.all\_company\_ids))union allselect \* from (selectc2.company\_name as company\_name,c1.all\_company\_ids as all\_company\_idsfrom {{ ref(‘companies\_merge\_list’) }} mjoin companies\_pre\_merged c1 on m.old\_company\_id in UNNEST(c1.all\_company\_ids)join companies\_pre\_merged c2 on m.company\_id in UNNEST(c2.all\_company\_ids)))group by 1)) mon c.company\_name = m.company\_namewhere c.company\_name not in (selectc2.company\_namefrom {{ ref(‘companies\_merge\_list’) }} mjoin companies\_pre\_merged c2 on m.old\_company\_id in UNNEST(c2.all\_company\_ids))) The process is the same for Snowflake data warehouse targets albeit with slightly different SQL to reflect the different approach to repeating nested columns that this platform uses left outer join (select company\_name, array\_agg(all\_company\_ids) as all\_company\_idsfrom (selectc2.company\_name as company\_name,c2.all\_company\_ids as all\_company\_idsfrom {{ ref(‘companies\_merge\_list’) }} mjoin (SELECT c1.company\_name, c1f.value::string as all\_company\_ids from {{ ref(‘int\_companies\_pre\_merged’) }} c1,table(flatten(c1.all\_company\_ids)) c1f) c1on m.old\_company\_id = c1.all\_company\_idsjoin (SELECT c2.company\_name, c2f.value::string as all\_company\_ids from {{ ref(‘int\_companies\_pre\_merged’) }} c2,table(flatten(c2.all\_company\_ids)) c2f) c2on m.company\_id = c2.all\_company\_idsunion allselectc2.company\_name as company\_name,c1.all\_company\_ids as all\_company\_idsfrom {{ ref(‘companies\_merge\_list’) }} mjoin (SELECT c1.company\_name, c1f.value::string as all\_company\_ids from {{ ref(‘int\_companies\_pre\_merged’) }} c1,table(flatten(c1.all\_company\_ids)) c1f) c1on m.old\_company\_id = c1.all\_company\_idsjoin (SELECT c2.company\_name, c2f.value::string as all\_company\_ids from {{ ref(‘int\_companies\_pre\_merged’) }} c2,table(flatten(c2.all\_company\_ids)) c2f) c2on m.company\_id = c2.all\_company\_ids)group by 1) mon c.company\_name = m.company\_namewhere c.company\_name not in (selectc2.company\_namefrom {{ ref(‘companies\_merge\_list’) }} mjoin (SELECT c2.company\_name, c2f.value::string as all\_company\_idsfrom {{ ref(‘int\_companies\_pre\_merged’) }} c2,table(flatten(c2.all\_company\_ids)) c2f) c2on m.old\_company\_id = c2.all\_company\_ids) The deduplicated and grouped list of companies together with their multi-valued source IDs fields and other field groups, suitably manually and additionally merged using our manual merge lookup file are then joined together ready for loading into the companies warehouse dimension table. SELECT i.all\_user\_ids,u.\*,e.all\_user\_emailsFROM (SELECT user\_name,MAX(contact\_is\_contractor) as contact\_is\_contractor,MAX(contact\_is\_staff) as contact\_is\_staff,MAX(contact\_weekly\_capacity) as contact\_weekly\_capacity ,MAX(user\_phone) as user\_phone,MAX(contact\_default\_hourly\_rate) as contact\_default\_hourly\_rate,MAX(contact\_cost\_rate) as contact\_cost\_rate,MAX(contact\_is\_active) as contact\_is\_active,MAX(user\_created\_ts) as user\_created\_ts,MAX(user\_last\_modified\_ts) as user\_last\_modified\_ts,FROM t\_users\_merge\_listGROUP BY 1) uJOIN user\_emails eON u.user\_name = COALESCE(e.user\_name,’Unknown’)JOIN user\_ids iON u.user\_name = i.user\_name ## Step 6: Generate Surrogate key for Dimension Tab;e Then coming along at the last minute but taking all the glory, the [warehouse dimension table model](https://github.com/rittmananalytics/ra_data_warehouse/blob/master/models/warehouse/w_crm/wh_companies_dim.sql) step at the end adds a unique identifier for the company record using the dbt\_utils.surrogate\_key macro. WITH companies\_dim as (SELECT{{ dbt\_utils.surrogate\_key(\[‘company\_name’\]) }} as company\_pk,\*FROM{{ ref(‘int\_companies’) }} c)select \* from companies\_dim ## Step 7: Un-nest the Source ID Arrays for Fact Joins Finally, when populating the fact tables that join to this dimension you’ll need a way to join to those repeating and variant data type columns that contain the array of source system IDs for each company, as shown in the screenshot below for Google BigQuery target warehouses. ![](https://rittmananalytics.com/api/public/og-image?src=https%3A%2F%2Fi0.wp.com%2Fnfqvbiqopvwmhnrrabjr.supabase.co%2Fstorage%2Fv1%2Fobject%2Fpublic%2Fblog-images%2Fimported-images%2F1752872505988-8oz0fc.png%3Fresize%3D1400%252C795%26ssl%3D1) array.png To do this when working with BigQuery as the target dimension we use the UNNEST() Standard SQL function to flatten those nested values and then perform the required fact table source to dimension table join, like [this example](https://github.com/rittmananalytics/ra_data_warehouse/blob/master/models/warehouse/w_projects/wh_delivery_projects_dim.sql) for our delivery projects fact table: WITH delivery\_projects AS(SELECT \*FROM {{ ref(‘int\_delivery\_projects’) }}),companies\_dim as (SELECT {{ dbt\_utils.star(from=ref(‘wh\_companies\_dim’)) }}from {{ ref(‘wh\_companies\_dim’) }})SELECT…FROMdelivery\_projects pJOIN companies\_dim cON p.company\_id IN UNNEST(c.all\_company\_ids) Snowflake performs the flatten of the repeated source system ID field using its own syntax, so that the equivalent SQL looks like this: companies\_dim as (SELECT c.company\_pk, cf.value::string as company\_idfrom {{ ref(‘wh\_companies\_dim’) }} c,table(flatten(c.all\_company\_ids)) cf)SELECT…FROMdelivery\_projects pJOIN companies\_dim cON p.company\_id = c.company\_id ## Interested? Find out More You can read more about our work with dbt, Google BigQuery, Snowflake and other modern data stack technologies on our website and blog: - [Rittman Analytics](https://rittmananalytics.com/home-index) homepage - [Data Centralization](https://rittmananalytics.com/data-centralization) consulting packages - [Rittman Analytics Blog dbt Posts](https://rittmananalytics.com/blog/tag/dbt) - [Contact Us](https://rittmananalytics.com/home-index/#about-us) How we handle deduplication and merging of company and contact data is just one of the design patterns we’ve included in our RA Warehouse for dbt framework that we’ve made available as free to use, open-source code in a public repo on our Github site. We’re a [dbt Preferred Consulting Partner](https://www.getdbt.com/ecosystem/) and this is one of the many ways we try and contribute value back to the dbt ecosystem, and we welcome issue reports, pull requests and suggestions on other problems encountered on data centralization projects that it’d be cool for us to solve using this framework. Or are you interested but don’t have the time to do it yourself? No problem! [Click here](https://calendly.com/markrittman/initial-discovery-call-with-mark-rittman) to book a 100% free, no-obligation 30 minute call to discuss your data needs and how we could help get your data centralization initiative moving now. [Data Engineering](https://rittmananalytics.com/blog/tag/Data+Engineering)[Data Warehousing](https://rittmananalytics.com/blog/tag/Data+Warehousing)[dbt](https://rittmananalytics.com/blog/tag/dbt) [Mark Rittman](https://rittmananalytics.com/blog?author=63c478ac6811232b3153d585)