Jak budować odporny pipeline importu CSV do PostgreSQL

Dlaczego import CSV wymaga procesu, a nie tylko COPY

Import plików CSV do PostgreSQL na pierwszy rzut oka wydaje się prosty: pobrać plik, wczytać dane i gotowe. W praktyce to właśnie na etapie importu najczęściej pojawiają się problemy z jakością danych, niespójnościami formatów i trudnymi do odtworzenia błędami. W małych firmach CSV bywa nadal podstawowym formatem wymiany danych z księgowością, ERP, handlowcami czy zewnętrznymi dostawcami.

Jeśli podejdziesz do importu jak do jednego kroku, szybko dostaniesz mieszankę rekordów częściowo poprawnych, częściowo uszkodzonych i trudnych do późniejszego zdiagnozowania. Lepsze podejście to podział na etapy: staging, walidację, normalizację, dopiero potem zapis do tabel docelowych.

Typowy przepływ importu

Dobrze zaprojektowany pipeline importu CSV zwykle składa się z kilku kroków:

• odczyt pliku i wykrycie formatu,
• zapis surowych danych do tabeli tymczasowej lub pomocniczej,
• walidacja struktury i typów,
• normalizacja wartości tekstowych i dat,
• deduplikacja,
• dopiero na końcu zapis do tabel produkcyjnych.

Taki układ daje trzy ważne korzyści: łatwiej odtworzyć błąd, łatwiej poprawić dane bez utraty surowego pliku i łatwiej rozdzielić odpowiedzialność między warstwy aplikacji.

Staging jako bezpieczny bufor

W praktyce warto trzymać surowy import w osobnej tabeli, np. import_csv_raw. Pole może odpowiadać jednej kolumnie z pliku albo przechowywać cały wiersz jako tekst lub jsonb, zależnie od potrzeb. Najważniejsze jest to, by nie mieszać danych wejściowych z danymi już zweryfikowanymi.

W PostgreSQL dobrym wzorcem jest też dodanie pól technicznych:

• import_batch_id do identyfikacji paczki,
• source_file_name do śledzenia pochodzenia,
• row_number do wskazania numeru wiersza z pliku,
• validation_status do oznaczenia wyniku walidacji,
• validation_errors do przechowania komunikatów.

Dzięki temu można szybko odpowiedzieć na pytanie, które rekordy przeszły, a które trzeba poprawić.

Walidacja struktury i typów

CSV nie niesie ze sobą schematu, więc aplikacja musi sama sprawdzić, czy liczba kolumn się zgadza, czy nagłówki są oczekiwane i czy dane dają się zrzutować do właściwych typów. Najczęstsze problemy to:

• puste pola tam, gdzie oczekiwany jest NOT NULL,
• błędny format daty,
• liczby zapisane z przecinkiem zamiast kropki lub odwrotnie,
• nadmiarowe spacje,
• niejednolite separatory dziesiętne,
• złe kodowanie pliku.

Warto sprawdzać dane jeszcze zanim trafią do docelowych tabel, bo wtedy komunikat o błędzie może być bardziej precyzyjny niż standardowy wyjątek bazy.

import csv
from datetime import datetime

with open("import.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for i, row in enumerate(reader, start=2):
        if not row.get("nip"):
            print(f"Wiersz {i}: brak NIP")
        try:
            datetime.strptime(row["data_transakcji"], "%Y-%m-%d")
        except ValueError:
            print(f"Wiersz {i}: nieprawidłowa data")

Taki prosty etap walidacji nie zastępuje bazy danych, ale znacząco ogranicza liczbę błędów przy zapisie.

Normalizacja danych przed zapisem

Nawet poprawne dane często wymagają ujednolicenia. W praktyce chodzi o to, by różne warianty tego samego rekordu wyglądały tak samo w bazie. Dotyczy to zwłaszcza nazw, numerów identyfikacyjnych, adresów i pól kontaktowych.

Najczęstsze operacje normalizacyjne to:

• usunięcie zbędnych spacji na początku i końcu,
• sprowadzenie tekstu do jednego wariantu wielkości liter,
• zamiana wielokrotnych spacji na pojedynczą,
• ujednolicenie formatów dat,
• oczyszczenie numerów z separatorów, jeśli nie są potrzebne,
• rozdzielenie pól złożonych na osobne kolumny.

Warto pamiętać, że normalizacja nie powinna niszczyć oryginalnego sensu danych. Jeśli potrzebujesz pełnego brzmienia nazwy z dokumentu, zachowaj je w osobnym polu, a do porównań używaj wersji znormalizowanej.

Deduplikacja w ETL: reguły i priorytety

Duplikaty w CSV pojawiają się często: ten sam kontrahent może wystąpić w kilku plikach, w różnych zapisach lub z drobnymi różnicami w nazwie. Deduplikację najlepiej robić na podstawie reguł biznesowych, a nie tylko pełnego porównania tekstu.

Przykładowe reguły:

• identyfikacja po NIP, jeśli jest dostępny i poprawny,
• identyfikacja po parze regon + nazwa,
• identyfikacja po normalizowanej nazwie i adresie,
• dla transakcji: identyfikacja po numerze dokumentu i dacie.

Jeśli w pliku są dwa rekordy o tym samym kluczu biznesowym, trzeba ustalić, który jest nadrzędny. Często wygrywa rekord z bardziej kompletnymi danymi albo nowszą informacją. Ważne, by ta zasada była jawna i powtarzalna.

W PostgreSQL deduplikację można wspierać przez indeksy unikalne na kolumnach znormalizowanych. Poniższy przykład pokazuje prosty wariant z ON CONFLICT:

INSERT INTO kontrahenci (nip, nazwa, miasto)
VALUES ('1234567890', 'Przyklad Sp. z o.o.', 'Warszawa')
ON CONFLICT (nip)
DO UPDATE SET
  nazwa = EXCLUDED.nazwa,
  miasto = EXCLUDED.miasto;

To podejście jest przydatne, gdy NIP rzeczywiście jest stabilnym kluczem. Jeśli nie, lepiej oprzeć logikę na bardziej złożonym zestawie pól i osobnej warstwie dopasowania.

Walidacja po stronie PostgreSQL

Część reguł warto przenieść do samej bazy. Dzięki temu nawet jeśli aplikacja importująca zawiedzie, baza nadal obroni spójność danych. Przydatne mechanizmy to:

• NOT NULL dla obowiązkowych pól,
• CHECK dla prostych reguł formatu,
• UNIQUE dla kluczy biznesowych,
• klucze obce dla relacji między tabelami,
• typy danych dopasowane do znaczenia kolumny.

To szczególnie ważne przy danych, które później zasilają raporty, integracje albo automatyczne rozliczenia. Im mniej „luźnych” rekordów trafi do produkcji, tym mniej pracy przy naprawach.

Obsługa błędów i raportowanie

Dobry import nie kończy się na sukcesie technicznym. Użytkownik powinien dostać jasną informację, co się udało, a co wymaga poprawy. Warto generować raport z podziałem na:

• rekordy poprawne,
• rekordy odrzucone,
• rekordy poprawne po automatycznej normalizacji,
• rekordy wymagające ręcznej decyzji.

Taki raport jest szczególnie przydatny, gdy CSV pochodzi od kilku osób lub systemów. Dzięki temu można szybko wskazać źródło błędów i poprawić proces u źródła, zamiast ręcznie naprawiać każdy kolejny plik.

Praktyczna zasada: najpierw jakość, potem automatyzacja

Automatyzacja importu ma sens dopiero wtedy, gdy znasz typowe błędy w danych i masz na nie przygotowaną reakcję. Jeśli plik bywa niestabilny, warto zacząć od półautomatycznego procesu: walidacja, raport, akceptacja, dopiero potem zapis.

Najlepsze pipeline’y importowe mają kilka cech wspólnych:

• są idempotentne, czyli można je uruchomić ponownie bez tworzenia chaosu,
• zachowują surowe dane,
• rozdzielają walidację od zapisu,
• mają czytelne logi,
• pozwalają odtworzyć historię importu.

Podsumowanie

Odporny import CSV do PostgreSQL to nie pojedyncze COPY, lecz proces złożony z kilku warstw. Surowe dane warto najpierw odłożyć do stagingu, potem sprawdzić ich strukturę, znormalizować, zdeduplikować i dopiero na końcu zapisać do tabel docelowych. Taki model zmniejsza liczbę błędów, poprawia audytowalność i ułatwia integrację z innymi systemami.

Jeśli chcesz budować podobne procesy szybciej, możesz skorzystać z API dla programistów albo z bezpłatnych narzędzi.