wave Hello In my org we have a classic Airbyte dbt architec Airbyte #replication-troubleshooting

:wave: Hello In my org we have a classic Airbyte/d...

Massy Bourennani

01/23/2023, 6:32 PM

👋 Hello In my org we have a classic Airbyte/dbt architecture. When it comes to materialization of incremental models we are discussing how we fetch new rows from airbyte raw tables to construct fact/dim tables. We are considering two definitions: 1. new rows according to

airbyte_emitted_at

column 2. new rows according to a

cursor_column

most of the time it’s an

updated_at

(a column that airbyte would typically use to sync streams incrementally) wanted to know how everyone was tackling this. Example in thread 🧵

➕ 4

Massy Bourennani

01/23/2023, 6:32 PM

Say we have a workday connector and we are currently fetching employee data using an

incremental sync - append

. This stream has a state defined by

updated_at

field of the Employee record. It represents the las time a record has been updated (name changed, department changed….). In the dbt layer we have an

employee

table that is construct in an incremental fashion. To do so we have two choices:

Massy Bourennani

01/23/2023, 6:33 PM

Copy code

-- in model workday_employee.sql
-- the only difference is in the {% if is_incremental() %} block

-- option 1: Using airbyte_emitted_at column
WITH new_updates AS (
SELECT *, 
       _airbyte_emitted_at as airbyte_emitted_timestamp
FROM {{ source('airbyte_workday', '_airbyte_raw_incremental_workday_employee') }}

{% if is_incremental() %}

WHERE airbyte_emitted_timestamp >= (SELECT IFNULL(max(airbyte_emitted_timestamp), TO_TIMESTAMP(0)) FROM {{this}})

{% endif %}
)

SELECT *,
       airbyte_emitted_timestamp
FROM new_updates

Massy Bourennani

01/23/2023, 6:40 PM

Copy code

-- option 2: Using cursor column
WITH new_updates AS (
SELECT *, 
       _airbyte_data:updated_at::timestamp_tz as updated_at --cursor column
FROM {{ source('airbyte_workday', '_airbyte_raw_incremental_workday_employee') }}

{% if is_incremental() %}

WHERE updated_at >= (SELECT IFNULL(max(updated_at), TO_TIMESTAMP(0)) FROM {{this}})

{% endif %}
)

SELECT *,
       updated_at
FROM new_updates

Marcos Marx (Airbyte)

01/23/2023, 10:23 PM

The sync mode incremental + dedup sync mode uses the cursor column to deduplicate and retrieve latest value.

Massy Bourennani

01/24/2023, 1:09 PM

ok that makes sense

Massy Bourennani

01/24/2023, 1:09 PM

thanks for the answer Marcos 🙂

Adrian Bakula

01/24/2023, 3:55 PM

@Marcos Marx (Airbyte) this is true on the airbyte side. Do you have any suggestions on which we should be using on the dbt side? Between

airbyte_emitted_at

cursor_column

user

01/24/2023, 5:46 PM

Probably

cursor_column

first because is linked to the real record updates, if it isn't stable you can use

airbyte_emitted_at

5 Views

Open in Slack

Previous Next