How to do incremental deduplication in ETL: step by step

Learn how to implement incremental deduplication in ETL to load only new records without duplicating existing data — useful when you receive periodic files or APIs and want to ensure integrity and efficiency. This example uses Python to show the why and how in a practical way.

Prerequisites

• Python 3.8+ installed
• Libraries: pandas, sqlalchemy, sqlite3 (or equivalent)
• Basic knowledge of SQL and pandas
• Text editor and terminal

Step 1: Understand the problem and choose the strategy

Incremental deduplication avoids reloading everything and redoing expensive comparisons. Common strategies: unique key (surrogate key/business key), row hash, or timestamp comparison. We will use a business key and a row hash to detect changes. This allows identifying new, changed, or unchanged records.

Step 2: Prepare a minimal example (CSV file)

Create a small CSV file that represents periodic sales data. Have a business key called order_id.

# vendas_1.csv
order_id,customer,amount,date
1,Ana,100,2026-01-01
2,Bruno,50,2026-01-02
3,Carlos,75,2026-01-03

# vendas_2.csv (next delivery, with duplicates and change)
order_id,customer,amount,date
2,Bruno,50,2026-01-02
3,Carlos,80,2026-01-03
4,Daniela,60,2026-01-04

Step 3: Write the basic ETL code with incremental deduplication

Explanation: we will read the new file, compute a hash per row, compare with the destination table in SQLite and insert only new or changed records. The example is minimal and functional.

import pandas as pd
import hashlib
from sqlalchemy import create_engine, text

# Função para criar hash de uma linha (exclui order_id)
def row_hash(row):
    payload = '|'.join(str(row[c]) for c in sorted(row.index) if c != 'order_id')
    return hashlib.md5(payload.encode('utf-8')).hexdigest()

# Engine SQLite (trocar por SQL Server/Postgres com connection string apropriada)
engine = create_engine('sqlite:///etl_example.db')

# Ler novo ficheiro
df_new = pd.read_csv('vendas_2.csv')
# Calcular hash
df_new['row_hash'] = df_new.apply(row_hash, axis=1)

# Garantir tabela destino existe com colunas: order_id, customer, amount, date, row_hash
with engine.begin() as conn:
    conn.execute(text('''
        CREATE TABLE IF NOT EXISTS vendas (
            order_id INTEGER PRIMARY KEY,
            customer TEXT,
            amount REAL,
            date TEXT,
            row_hash TEXT
        )
    '''))

# Carregar hashes existentes
with engine.connect() as conn:
    existing = pd.read_sql('SELECT order_id, row_hash FROM vendas', conn)

# Determinar quais inserir ou actualizar
merged = df_new.merge(existing, on='order_id', how='left', suffixes=('', '_existing'))
# Novo se row_hash_existing for nulo; alterado se diferente
to_insert = merged[merged['row_hash_existing'].isna()]
to_update = merged[(~merged['row_hash_existing'].isna()) & (merged['row_hash_existing'] != merged['row_hash'])]

# Inserir novos
if not to_insert.empty:
    to_insert[['order_id','customer','amount','date','row_hash']].to_sql('vendas', engine, if_exists='append', index=False)

# Actualizar alterados (simples: UPDATE por order_id)
with engine.begin() as conn:
    for _, row in to_update.iterrows():
        conn.execute(text('''
            UPDATE vendas SET customer = :customer, amount = :amount, date = :date, row_hash = :row_hash
            WHERE order_id = :order_id
        '''), {
            'customer': row['customer'], 'amount': row['amount'], 'date': row['date'], 'row_hash': row['row_hash'], 'order_id': int(row['order_id'])
        })

print('Novos:', len(to_insert), 'Alterados:', len(to_update))

Step 4: Handle common errors and ensure idempotency

Common errors: key collisions, inconsistent date formats, and partially loaded files. Best practices: use transactions (create_engine().begin() already helps), validate schema with pandas dtypes, and generate simple logs. Idempotency here results from using the business key and comparison by hash.

# Exemplo simples de validação de schema antes de inserir
expected_cols = {'order_id','customer','amount','date'}
if not expected_cols.issubset(set(df_new.columns)):
    raise ValueError('Faltam colunas no ficheiro novo')

# Normalizar tipos
df_new['order_id'] = df_new['order_id'].astype(int)

Verify the result

To confirm: query the destination table and validate the records. Expect to see 4 records after loading vendas_1.csv and vendas_2.csv, with order_id 3 updated to amount 80.

from sqlalchemy import create_engine
import pandas as pd
engine = create_engine('sqlite:///etl_example.db')
print(pd.read_sql('SELECT * FROM vendas ORDER BY order_id', engine))

Conclusion

Incremental deduplication using a business key + row hash is a simple and efficient technique for daily/periodic ETL. Next steps: adapt for databases like SQL Server/Postgres, add structured logging, and optimize batch updates. Tip: what is the best business key for your data?