Budowanie pipeline'u danych z Cursorem

Cursor buduje wiarygodny pipeline danych wtedy, gdy Plan Mode najpierw zapisuje kontrakt (schematy każdego etapu, klucze naturalne, znaczniki przyrostowe i niezmienniki), a Agent implementuje ekstrakcję, transformację i ładowanie pod testy na fixture’ach, których nie wolno mu edytować. Poprawności danych dowodzą testy ponownego uruchomienia, uzgadnianie liczby wierszy i sum ze źródłem oraz kontrole jakości po załadowaniu, a nie czytanie wygenerowanego kodu.

Ten przewodnik jest dla programistów i inżynierów danych, którzy budują pipeline’y w Cursorze, oraz dla tech leadów, którzy muszą podpisać się pod liczbami. Sytuacja: dział analiz potrzebuje codziennego raportu z repliki Postgresa, API Stripe’a i plików CSV, które system magazynowy wrzuca do S3. Dziś robi to skrypt na 400 linii. Po cichu przestaje działać, gdy ktoś zmieni nazwę kolumny w CSV, dubluje wiersze po ręcznym ponownym uruchomieniu, a o 3 w nocy nikt nie wie, na którym etapie stanął. Agent szybko napisze następcę. Sam z siebie nie wie jednak, że inner join, który po cichu gubi 2% zamówień, to błąd, i właśnie o to, jak to wykryć, chodzi w tym tekście.

Co zyskasz, budując pipeline w Cursorze

Prompt dla Plan Mode, który zamienia wymagania w plik z kontraktem pipeline’u, pod który Agent implementuje kod.
Regułę projektu, dzięki której idempotentność, pliki kwarantanny i ustrukturyzowane błędy są domyślne w każdym etapie.
Dostęp Agenta do bazy tylko do odczytu przez serwer MCP dla Postgresa, więc schematy pochodzą ze źródła, a nie z domysłów.
Sześć promptów do skopiowania: ekstraktory, transformacje pisane test-first, idempotentny loader, orkiestracja i kontrole jakości.
Drabinę weryfikacji, która dowodzi poprawności przebiegu liczbami i sumami, oraz tabelę awarii z krokami naprawczymi.

Dlaczego pipeline potrzebuje kontraktu przed kodem?

Agent szybko generuje wiarygodnie wyglądający kod transformacji i właśnie w tym tkwi problem: błędny pipeline też tworzy tabele, a dashboardy czytają je bez słowa skargi. Kontrakt jest wyrocznią. Dla każdego etapu określa schemat wejścia i wyjścia, klucz naturalny, znacznik przyrostowy (watermark) i niezmienniki, które muszą być spełnione, na przykład „każde wyekstrahowane zamówienie zostaje załadowane albo trafia do kwarantanny”. Gdy niezmienniki istnieją w pliku, testy i kontrole mogą je egzekwować, a review sprowadza się do pytania, czy kontrakt jest dobry, a nie czy 2000 linii TypeScriptu wygląda dobrze. Ten przewodnik stosuje zasadę weryfikowania dowodów zamiast czytania diffów.

Krok 1: Napisz kontrakt pipeline’u w Plan Mode

Plan Mode tworzy plan implementacji, zanim Agent napisze jakikolwiek kod (zob. tryby Agenta w szczegółach). Użyj go do przygotowania kontraktu, a potem przejrzyj i zacommituj ten plik, zanim powstanie pierwsza linia kodu pipeline’u.

Plan a daily batch pipeline and write the result to docs/pipeline-contract.md.
Do not write any pipeline code yet.

Sources:
- Postgres read replica: orders, order_items, customers, products (use updated_at for incremental reads)
- Stripe API: payment intents, refunds, disputes for the run date
- CSV inventory snapshots in s3://acme-warehouse-exports/inventory/YYYY-MM-DD/

Destination: Snowflake, schema ANALYTICS, star schema (fact_orders, dim_customer, dim_product, fact_inventory).

Requirements:
- Runs daily at 02:00 UTC for the previous UTC day; supports backfill of a date range.
- Idempotent: re-running any date produces identical tables, never duplicates.
- Partial failure: if Stripe is down, load the other sources and mark the run "partial".
- Every input row ends up loaded or quarantined with a reason. Nothing disappears.

For each stage (extract per source, transform, load, quality checks) write:
1. Input and output schema with types and nullability.
2. Natural key and deduplication rule.
3. Watermark or partition used for incremental and re-run behaviour.
4. Invariants a test can check (row conservation, sum conservation of amounts, key uniqueness).
5. Failure behaviour: retry, skip-and-mark-partial, or abort.

Before writing the file, ask me about anything you would otherwise have to assume:
volumes, late-arriving data, currency handling, and timezone of the source timestamps.

Czytaj kontrakt, nie kod, który z niego powstanie. Wszystkie pytania warte czasu człowieka są właśnie w nim: czy order_id jest naprawdę unikalny między kolejnymi ekstrakcjami, czy kwoty ze Stripe’a są w najmniejszej jednostce waluty i czy „poprzedni dzień” oznacza UTC, czy strefę czasową biznesu. Zła odpowiedź tutaj zamienia się później w zły dashboard, a poprawić ją w 150-linijkowym pliku Markdown jest dużo taniej.

Krok 2: Dodaj regułę projektu dla kodu pipeline’u

Reguła projektu przenosi konwencje do każdej sesji Agenta, więc piąty ekstraktor powstaje według tego samego wzorca co pierwszy. Ogranicz ją do katalogów pipeline’u i trzymaj krótko; format pliku i opcje zakresu opisuje strona o regułach projektu.

Pipeline code in src/extractors, src/transformers, src/loaders, src/quality:

- The contract in docs/pipeline-contract.md is the source of truth. If code and contract
  disagree, stop and ask; do not change the contract to fit the code.
- Every stage is idempotent for a given run_date. Loads use MERGE on the natural key, never INSERT.
- Validate every external payload with a Zod schema at the boundary. Unknown fields are logged
  and dropped, not fatal. Missing required fields send the row to the quarantine file with a reason.
- Every stage returns { rowsIn, rowsOut, rowsQuarantined } and asserts rowsIn = rowsOut + rowsQuarantined.
- Every error log includes run_id, stage, source, run_date and the offending key.
- Timestamps are handled in UTC; convert only in the presentation layer.
- Never edit files under tests/fixtures or tests/expected. If a fixture test fails, fix the code
  or report why the expected output is wrong.

Ostatnia linia znaczy więcej, niż się wydaje. Agent, który ma sprawić, żeby test przeszedł, czasem „naprawia” oczekiwany wynik zamiast kodu. Pełny zestaw zabezpieczeń opisuje strona o ochronie wyroczni testowej; dla pipeline’u minimum to ta reguła plus sprawdzenie w CI, które nie przepuszcza pull requesta zmieniającego fixture razem z kodem, który ten fixture testuje. Oto takie sprawdzenie jako workflow GitHub Actions. Potrzebuje pełnej historii (fetch-depth: 0), żeby porównać zmiany z gałęzią bazową, i wystarcza mu dostęp do repozytorium tylko do odczytu:

name: fixture-guard
on: pull_request
permissions:
  contents: read
jobs:
  guard:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
        with:
          persist-credentials: false
          fetch-depth: 0
      - name: Fail when fixtures and pipeline code change together
        env:
          BASE_REF: ${{ github.base_ref }}
        run: |
          changed=$(git diff --name-only "origin/$BASE_REF...HEAD")
          if echo "$changed" | grep -q '^tests/\(fixtures\|expected\)/' && echo "$changed" | grep -q '^src/'; then
            echo "::error::Fixtures and code changed in one pull request. Split them."
            exit 1
          fi

Fixture, który naprawdę jest błędny, trafia wtedy do osobnego pull requesta, gdzie właściciel danych ocenia go bez zmian w kodzie, za którymi mogłaby się ukryć. Jeśli wolisz bramkę zatwierdzeń zamiast twardego błędu, dodaj do CODEOWNERS wpis tests/fixtures/ @data-owners i włącz dla gałęzi wymagane review właścicieli kodu. W obu przypadkach sam check nie chroni się przed manipulacją: przy pull_request GitHub uruchamia ten plik workflow z samego pull requesta, więc pull request może zmienić fixture-guard.yml i nadal przejść na zielono. Dodaj do CODEOWNERS wpis /.github/ @platform-owners z wymaganym review właścicieli kodu i ustaw fixture-guard jako wymagany status check, żeby zmiana strażnika nie weszła bez zgody jego właściciela.

Krok 3: Daj Agentowi dostęp do schematu źródła tylko do odczytu

Agent pisze lepsze ekstraktory, gdy widzi prawdziwe typy kolumn, nullowalność i indeksy, zamiast zgadywać je ze starego kodu. Umożliwia to Postgres MCP Pro (PyPI postgres-mcp, wersja 0.3.0, ostatnie wydanie 2025-05-16, sprawdzone 2026-09-26), a --access-mode=restricted ogranicza go do transakcji tylko do odczytu z limitami czasu wykonania. Od ponad roku nie ma nowego wydania, więc przypnij wersję, a jeśli potrzebujesz aktywnie rozwijanego serwera, wybierz DBHub albo MCP Toolbox ze strony Serwery MCP dla baz danych. To pakiet Pythona, więc uruchamiasz go przez uvx (wymaga uv), nie npx. Wpis umieść w globalnym ~/.cursor/mcp.json, a nie w projektowym .cursor/mcp.json, żeby connection string nie trafił do repozytorium:

{
  "mcpServers": {
    "postgres": {
      "command": "uvx",
      "args": ["postgres-mcp==0.3.0", "--access-mode=restricted"],
      "env": { "DATABASE_URI": "postgresql://agent_ro@replica.internal:5432/app" }
    }
  }
}

Hasło roli agent_ro trzymaj w ~/.pgpass, a nie w pliku JSON, i nadaj tej roli wyłącznie SELECT. Kieruj ją na replikę albo kopię stagingową, nigdy na bazę główną: wiersze odczytane przez agenta trafiają do modelu, a kolumna z dowolnym tekstem może zawierać instrukcje. Serwery MCP dla baz danych porównują alternatywy (DBHub, MCP Toolbox, Supabase, Neon), a konfiguracja MCP w Cursorze opisuje włączanie i debugowanie serwera.

Krok 4: Zbuduj ekstraktory z kwarantanną zamiast cichego gubienia wierszy

Każdy ekstraktor daje ten sam wynik: pliki JSONL w katalogu stagingowym, plik kwarantanny i rekord z licznikami. Buduj jeden ekstraktor na sesję, żeby każdy diff dało się zweryfikować jego testami.

Implement the Postgres and Stripe extractors described in docs/pipeline-contract.md.

Postgres (src/extractors/postgres.ts):
- Use the postgres MCP server to read the actual column types of orders, order_items,
  customers and products before writing any types. Generate TypeScript row types from them.
- Use pg with a Pool. Read rows where updated_at is in [watermark, run_date + 1 day).
- Store the watermark per table in the pipeline_state table only after the stage succeeds.

Stripe (src/extractors/stripe.ts):
- Use the stripe SDK's auto-pagination for payment intents, refunds and disputes created in the run date.
- Retry 429 and 5xx with exponential backoff and jitter, max 5 attempts. If Stripe stays
  unavailable, return status "skipped" with the error instead of throwing.

Both extractors:
- Validate every row with a Zod schema. Invalid rows go to staging/quarantine/<source>.jsonl
  with the validation error; valid rows go to staging/<source>/<table>.jsonl.
- Return { rowsIn, rowsOut, rowsQuarantined } and assert rowsIn = rowsOut + rowsQuarantined.

Write Vitest tests with recorded fixtures (no network): a page boundary in Stripe pagination,
a 429 followed by success, a row with a null required field, and a re-run with an unchanged watermark
that must extract zero new rows. Run the tests and show me the output.

Źródła CSV zasługują na osobny prompt, bo zmieniają się bez zapowiedzi. Poproś Agenta, żeby mapował nazwy nagłówków przez plik konfiguracyjny (config/csv-columns.json), dzięki czemu zmiana nazwy kolumny to jedna linia konfiguracji; żeby brak dziennego pliku traktował jako „brak danych”, a nie błąd, jeśli kontrakt mówi, że weekendy są pomijane; i żeby parsował przez csv-parse, tak by pola w cudzysłowach z nowymi liniami przetrwały. Reguła kwarantanny z kroku 2 obowiązuje bez zmian.

Krok 5: Napisz testy transformacji przed transformacjami

To w transformacjach dane giną po cichu: join po kluczu, który bywa nullem, deduplikacja zostawiająca starszy wiersz, przeliczenie centów na dolary wykonane dwa razy. Każ Agentowi najpierw napisać padające testy na podstawie kontraktu, a dopiero potem kod.

Read docs/pipeline-contract.md, section "transform". Do not write src/transformers/orders.ts yet.

First, write tests in tests/transformers/orders.test.ts with small hand-written fixtures in
tests/fixtures/orders/ covering:
1. An order with no matching customer (customer name "Unknown", email null, row kept).
2. An order with no Stripe payment (payment_status "pending", row kept).
3. The same order_id in today's and yesterday's extract (keep the row with the later updated_at).
4. Amounts in cents converted to dollars exactly once, using integer arithmetic until the final value.
5. An order whose updated_at is 23:59:59.999 UTC lands in the correct run_date.

Then add property-based tests with fast-check over generated inputs:
- Row conservation: output rows + quarantined rows = distinct input order_ids.
- Sum conservation: the sum of amount_cents in equals the sum out plus the sum quarantined.
- Order independence: shuffling the input rows does not change the output.

Run the tests, confirm they fail for the right reason, then implement the transformer until they pass.
Do not modify anything under tests/fixtures after this point.

Najwięcej pracy wykonują testy oparte na własnościach, bo sprawdzają niezmiennik na tysiącach wygenerowanych danych wejściowych, a nie tylko na pięciu przypadkach, które przyszły ci do głowy. Testy oparte na własnościach z agentami wchodzą głębiej w dobór własności, a zarządzanie danymi testowymi pokazuje, jak budować realistyczne fixture’y bez kopiowania wierszy z produkcji.

Krok 6: Zrób idempotentny loader i udowodnij to, uruchamiając go dwa razy

Implement src/loaders/snowflake.ts per docs/pipeline-contract.md, section "load".

- Use snowflake-sdk. Load each JSONL file into a temporary staging table in 10,000-row batches.
- Then run one MERGE per table from the staging table into the target, on the natural key,
  inside a transaction: update when the source updated_at is newer, insert when missing.
  On failure roll back and log run_id, table and the error.
- Write a pipeline_runs row: run_id, run_date, status, and rowsIn/rowsOut/rowsQuarantined per stage.
- Support --dry-run: validate and report counts, write nothing.

Add an integration test that runs against a disposable schema named PIPELINE_TEST_<random suffix>:
load the fixture day, record row counts and SUM(amount_usd) per table, load the same day again,
and assert both numbers are unchanged. Drop the schema at the end of the test.
Read Snowflake credentials from environment variables only.

Test podwójnego ładowania to najcenniejszy test w całym pipeline’ie. INSERT w miejscu, gdzie powinien być MERGE, przechodzi każdą inną kontrolę przy pierwszym uruchomieniu i podwaja przychody przy drugim.

Krok 7: Zorkiestruj przebieg i sprawdź dane po załadowaniu

Implement src/orchestrator.ts and src/quality/checks.ts per docs/pipeline-contract.md.

Orchestrator:
- CLI: npx tsx src/orchestrator.ts --date 2026-09-25, or --from and --to for a backfill (one date at a time).
- Run the three extractors concurrently, then transform, then load.
- A skipped source marks the run "partial" and records which source to retry; any other failure aborts
  the run before loading and leaves the previous day's tables untouched.
- Take a lock row in pipeline_runs for run_date so two runs for the same date cannot overlap.

Quality checks (run after load, results written to pipeline_quality):
- Reconciliation: orders loaded for run_date equals orders in the Postgres replica for run_date
  minus quarantined rows; the same for SUM(amount_cents) against Stripe.
- Uniqueness of natural keys in every fact table.
- Referential integrity: every fact_orders.customer_key exists in dim_customer.
- Volume: row count within the range of the previous 28 days' min and max; flag, do not fail.
- Freshness: newest updated_at in fact_orders is within 26 hours of the run time.
Failures post to the webhook in QUALITY_WEBHOOK_URL with run_id and the failing query.

Uzgadnianie ze źródłem (reconciliation) to kontrola, która uwiarygadnia całą resztę. Porównuje hurtownię ze źródłami liczbami, które każdy może przeliczyć sam, więc recenzent, który nigdy nie otworzy orders.ts, i tak może stwierdzić, czy dane z danego dnia są poprawne. Dashboardy i routing alertów na bazie tabeli pipeline_quality opisuje strona o monitoringu z Cursorem.

Jak zweryfikować pipeline bez czytania każdej linii?

Ułóż bramki tak, żeby każda łapała to, czego nie złapie poprzednia, i niech o merge’u decydują dowody, a nie diff.

Bramka	Czego dowodzi	Gdzie działa	Kto patrzy na wynik
Sprawdzanie typów i lint	Typy zgadzają się ze schematami odczytanymi ze źródła	CI przy każdym pull requeście	Nikt, dopóki nie jest czerwono
Testy na fixture’ach (krok 5)	Znane przypadki brzegowe działają zgodnie z kontraktem	CI	Autor, gdy test padnie
Testy oparte na własnościach (krok 5)	Zachowanie liczby wierszy i sum na wygenerowanych danych	CI	Autor, gdy test padnie
Test podwójnego ładowania (krok 6)	Ponowne uruchomienie nie dubluje ani nie zmienia danych	CI na jednorazowym schemacie	Autor, gdy test padnie
Odtworzenie prawdziwego dnia	Kod radzi sobie z danymi o kształcie produkcyjnym	Staging, przed merge’em	Recenzent czyta raport uzgodnienia
Kontrole po załadowaniu (krok 7)	Każdy produkcyjny przebieg zgadza się ze źródłami	Każde zaplanowane uruchomienie	Dyżurny, przez webhook

Na potrzeby odtworzenia uruchom nowy kod dla jednej niedawnej daty do schematu stagingowego i poproś Agenta o raport uzgodnienia: dla każdej tabeli liczba wierszy w źródle, załadowanych i w kwarantannie oraz sumy kwot, obok tych samych liczb z obecnego produkcyjnego pipeline’u. Dołącz raport do pull requesta. Ten raport, diff kontraktu i zielone CI tworzą pakiet dowodów (evidence bundle) dla zmiany. Właściciel danych zatwierdza kontrakt i liczby z uzgodnienia; recenzent potwierdza, że fixture’y nie zmieniły się razem z kodem.

Gdzie psują się pipeline’y zbudowane w Cursorze i jak to naprawić

Objaw	Prawdopodobna przyczyna	Naprawa
Załadowanych wierszy jest o kilka procent mniej niż w źródle, bez błędu	Inner join po kluczu, który bywa nullem, albo deduplikacja po złym kluczu	Znajdź zgubione klucze anti-joinem między stagingiem a tabelą docelową. Dodaj fixture z takim kluczem, popraw transformację i uruchom ponownie daty, których to dotyczy; `MERGE` sprawia, że to bezpieczne.
Przychody podwoiły się po ręcznym ponownym uruchomieniu	Loader używa `INSERT` albo klucz `MERGE` nie jest kluczem naturalnym	Usuń partycję run_date, popraw loader i uruchom ponownie. Przed merge’em poprawki dodaj test podwójnego ładowania.
Ekstraktor zaczyna padać dla jakiejś daty bez zmiany w kodzie	Dryf schematu źródła: nowa wartość enuma, zmieniona nazwa kolumny w CSV	Najpierw przeczytaj plik kwarantanny. Zaktualizuj schemat Zod albo `config/csv-columns.json` i uruchom tę datę ponownie.
Po ostatniej zmianie Agenta test zrobił się zielony, ale dane są złe	Agent zmienił fixture albo oczekiwany wynik pod kod	Uruchom `git diff --stat -- tests/fixtures tests/expected`, cofnij zmianę fixture’a i ponów prompt, wskazując padający test. Dodaj sprawdzenie w CI z kroku 2.
Zamówienia z późnego wieczoru lądują w złym dniu	Znaczniki czasu w źródle w czasie lokalnym, run_date w UTC	Ustal strefę czasową w kontrakcie, dodaj fixture 23:59:59.999 i przelicz ponownie zakres dat, którego dotyczy błąd.
Backfill dostaje 429 od Stripe’a albo spowalnia replikę	Backfill uruchamia daty równolegle	Uruchamiaj jedną datę naraz pętlą `--from`/`--to` orkiestratora i zachowaj ustawienia backoffu z kroku 4.
Awaria o 3 w nocy mówi tylko `Error: undefined`	Błędy rzucane bez kontekstu	Poproś Agenta o przegląd każdego `throw` i logu pod kątem reguły z kroku 2, a potem dodaj test, który wymusza awarię każdego etapu i sprawdza pola logu.

Dokąd dalej po zbudowaniu pipeline’u

Pracujesz w innym narzędziu? Tę samą pracę z bazą danych opisują przewodniki Praca z bazą danych w Claude Code i Praca z bazą danych w Codex, a praktykę niezależną od narzędzia — Projektowanie baz danych i zapytania.

Projektowanie bazy danych z Cursorem Zaprojektuj schematy hurtowni i źródeł, na których opiera się kontrakt pipeline'u.

Testowanie i TDD z Cursorem Przenieś pętlę test-first z kroku 5 na resztę kodu.

Monitoring z Cursorem Zamień kontrole po załadowaniu w dashboardy i alerty.

DevOps z Cursorem Wdróż i zaplanuj pipeline jako infrastrukturę w kodzie.

Projektowanie baz danych i zapytania (wszystkie narzędzia) Wspólny dla wszystkich narzędzi proces pracy nad schematami, migracjami i zapytaniami z agentem.

Receptury inżynierii danych dbt, potoki SQL, notatniki i testy danych z agentem.