dbt incremental model ap kreye duplicate records

Dyagnostik

Pwoblèm nan se filtè WHERE transaction_date >= MAX(transaction_date) la. Paske w ap itilize >= (pa >), chak fwa modèl la kouri, li reprann tout ranje ki gen menm transaction_date ak MAX() ki deja egziste nan tab la, epi li enjekte yo ankò. Kòm unique_key='transaction_id' sipoze fè yon MERGE/UPSERT, sa dwe evite duplikasyon si konfigirasyon merge a byen fèt — men rezon prensipal duplicate yo se ke plizyè transaction_id diferan ka gen menm transaction_date (menm jou/timestamp), e lojik >= la ap re-selectionne yo epi enjekte yo kòm nouvo ranje si merge/unique_key la pa aplike kòrèkteman, oswa si gen plizyè ranje sous ki gen menm transaction_id deja depi nan sous done a.

Bagay ki dwe verifye/eseye

1. Verifye si gen duplicate deja nan sous la (source('sales','transactions')) — si sous la deja gen plizyè liy pou menm transaction_id, unique_key a dbt p ap ka rezoud sa san yon dedup anvan (ex: QUALIFY ROW_NUMBER() OVER (PARTITION BY transaction_id ORDER BY transaction_date DESC) = 1).

2. Ranplase >= ak >, oswa pi bon toujou, itilize yon apwòch ki mwens riske pou double-processing:

{% if is_incremental() %}
WHERE transaction_date > (
    SELECT MAX(transaction_date)
    FROM {{ this }}
)
{% endif %}

3. Konfime incremental_strategy — sou Snowflake, si ou pa espesifye incremental_strategy, dbt itilize merge pa default kòm unique_key la defini, sa ki ta dwe fè yon upsert kòrèk. Men verifye nan dbt debug/logs si strategy a se byen merge epi ke unique_key la pa null nan okenn ranje (yon transaction_id NULL ka kreye pwoblèm nan MERGE la).

4. Ajoute yon dedup nan modèl la anvan chaje kòm mezi sekirite, menm si sous done a ta sipoze gen valè inik:

{{ config(
    materialized='incremental',
    unique_key='transaction_id',
    incremental_strategy='merge'
) }}
 
WITH source_data AS (
    SELECT
        transaction_id,
        customer_id,
        amount,
        transaction_date,
        ROW_NUMBER() OVER (
            PARTITION BY transaction_id
            ORDER BY transaction_date DESC
        ) AS rn
    FROM {{ source('sales', 'transactions') }}
    {% if is_incremental() %}
    WHERE transaction_date > (SELECT MAX(transaction_date) FROM {{ this }})
    {% endif %}
)
 
SELECT
    transaction_id,
    customer_id,
    amount,
    transaction_date
FROM source_data
WHERE rn = 1

Sa pou ou fè apre

• Kouri yon SELECT transaction_id, COUNT(*) FROM {{ source('sales','transactions') }} GROUP BY 1 HAVING COUNT(*) > 1 pou konfime si duplicate yo soti nan sous la oswa se rezilta filtè incremental la.
• Si tab final la deja gen duplicate, fòk ou fè yon dbt run --full-refresh apre ou fin korije lojik la, paske incremental merge a p ap otomatikman netwaye duplicate ki deja egziste.

Sa a se yon pwopozisyon — tanpri verifye rezilta requèt dyagnostik yo anvan ou aplike chanjman sa yo sou pwodiksyon.