Testování pipeline v Pythonu: pytest a Great Expectations (2026)

Aktualizováno: 22. června 2026

Testování datových pipeline v Pythonu znamená kombinovat pytest pro logiku transformací s nástroji typu Great Expectations nebo Pandera pro kontrolu kvality dat, které tečou pipeline. Bez obou vrstev se vám stane to, co se mi stalo už třikrát: pipeline projde všemi unit testy, ale v produkci spadne kvůli nullům ve sloupci, který před týdnem nikdo nehlídal. V tomhle průvodci si projdeme moderní stack pro rok 2026, ukážu reálné testy z mých pipeline a poradím, jak se vyhnout bolestivému backfillu.

• Datové pipeline potřebují dvě vrstvy testů: logické (pytest) a datově-kvalitní (Great Expectations, Pandera, dbt tests).
• Great Expectations 1.x (vydáno 2024) přineslo Fluent API a výrazně zjednodušilo integraci do Airflow/Dagster pipeline.
• Pandera 0.20+ podporuje schéma validaci pro pandas, Polars i PySpark s jedním DSL.
• Pytest fixtures s tmp_path a malými fixture datasety nahrazují drahé integrační testy nad celou produkční DB.
• Každá pipeline by měla mít test na idempotenci. Opětovné spuštění nesmí duplikovat řádky, a tohle je nejčastější zdroj backfill peklíčka.
• CI/CD běh musí pipeline testy spouštět při každé změně SQL nebo Python transformací, ne jen jednou týdně.

Proč testovat datové pipeline jinak než aplikace

Když píšete REST API, vstupy znáte. Schéma definuje request body a vy testujete, jak na něj odpoví kód. U datové pipeline je to obráceně. Kód se dost často nemění, mění se data: dodavatel přidá sloupec, marketing pošle CSV s desetinnou čárkou místo tečky, upstream API vrátí null tam, kde garantovalo string. Klasické unit testy tyhle situace nezachytí, protože testují kód s daty, která si sami vyrobíte.

V praxi to znamená, že potřebujete jak klasické testy logiky (pytest), tak průběžnou validaci dat (Great Expectations, Pandera, dbt tests). Tahle kombinace je jediné, co mi v posledních letech zachránilo víkend. Když pipeline spadne v 3:00, chci přesnou odpověď „chybí 12 % řádků ve sloupci customer_id", ne „something went wrong". Pokud teprve začínáte stavět pipeline a chcete pochopit i transformační vrstvu, doporučuju předtím přečíst průvodce ETL pipeline v Pythonu, který jsem psala dřív. Testy navazují přesně na to.

Tři vrstvy testování datové pipeline

Pipeline rozděluju vždycky na tři vrstvy a každá má svoje testy:

1. Unit testy logiky (pytest). Testují čisté Python/SQL funkce na malých fixture datasetech. Běhají v milisekundách, spouští se při každém commitu.
2. Data quality kontroly (Great Expectations / Pandera). Běží v rámci pipeline na reálných datech. Pokud expectace selže, pipeline se zastaví před tím, než špatná data zaplaví warehouse.
3. Integrační testy (dbt tests + end-to-end). Spouští mini-verzi pipeline proti testovacímu schématu. Validují, že transformace na sebe navazují a že referenční integrita drží.

Vrstva	Nástroj	Co testuje	Kdy běží
Unit	pytest	Logika transformací	Při každém commitu (CI)
Kvalita dat	Great Expectations, Pandera	Reálná data v pipeline	Po každém runu pipeline
Schéma	Pandera, dbt contracts	Typy a omezení sloupců	Před zápisem do warehouse
Integrace	dbt tests, pytest-dbt	Vztahy mezi modely	V CI a před produkčním runem

Pytest pro logiku transformací

Pytest se v datovém světě používá úplně stejně jako v aplikační vrstvě, jenom fixtures jsou trochu jiné. Honestly, typicky chci mít malý DataFrame uložený jako fixture a testovat transformaci jako čistou funkci. Tady je vzor, který používám v týmu:

# tests/conftest.py
import pandas as pd
import pytest

@pytest.fixture
def orders_raw():
    """Minimalni fixture imitujici surova data z upstream tabulky."""
    return pd.DataFrame({
        "order_id": [1, 2, 3, 4],
        "customer_id": [10, 10, 11, None],   # null musi byt explicitne testovany
        "amount_cents": [1500, 2300, 999, 5000],
        "created_at": pd.to_datetime([
            "2026-01-01", "2026-01-01",
            "2026-01-02", "2026-01-02",
        ]),
    })

# tests/test_transforms.py
from mypipeline.transforms import clean_orders

def test_clean_orders_drops_null_customers(orders_raw):
    result = clean_orders(orders_raw)
    assert result["customer_id"].isna().sum() == 0
    assert len(result) == 3  # jeden radek vypadne

def test_clean_orders_converts_cents_to_dollars(orders_raw):
    result = clean_orders(orders_raw)
    assert result["amount_usd"].iloc[0] == 15.00

Tři pravidla, která dodržujte:

• Žádné fixture na disku, pokud nemusíte. CSV soubory v repu se rozcházejí s realitou. Definujte data v kódu.
• Jeden test = jeden invariant. Nezkoušejte v jednom testu počet řádků a typ a obsah sloupce. Když to spadne, nevíte co.
• Parametrizujte hraniční případy přes @pytest.mark.parametrize: null, prázdný DataFrame, jeden řádek, milion řádků.

Great Expectations 1.x: kontrola kvality dat v produkci

Great Expectations (GX) je framework, kterým definujete „expectace", což jsou assertions o tom, jak data mají vypadat. Když data tyhle expectace splňují, pipeline pokračuje. Když ne, dostanete přesný report o tom, co se rozbilo. Verze 1.0 vyšla v roce 2024 a verze 1.3 z roku 2025 přidala Fluent API, které je výrazně přívětivější než staré YAML konfigurace.

Podrobnou specifikaci najdete v oficiální dokumentaci Great Expectations, ale pro start stačí tohle:

import great_expectations as gx
import pandas as pd

# Pipeline nacte data
df = pd.read_parquet("s3://bucket/orders/2026-06-22.parquet")

# Vytvorime ephemeral kontext (zadne YAML soubory)
context = gx.get_context(mode="ephemeral")

# Pripojime DataFrame jako data asset
data_source = context.data_sources.add_pandas("orders_source")
data_asset = data_source.add_dataframe_asset("daily_orders")
batch_def = data_asset.add_batch_definition_whole_dataframe("today")
batch = batch_def.get_batch(batch_parameters={"dataframe": df})

# Definujeme expectace
suite = context.suites.add(gx.ExpectationSuite("orders_quality"))
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount_cents", min_value=0, max_value=10_000_000
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)

# Validujeme
result = batch.validate(suite)
if not result.success:
    raise ValueError(f"Data quality check failed: {result}")

Pandera: schéma validace přímo v kódu

Pokud Great Expectations vnímáte jako těžkou váhu, Pandera je odpověď. Definujete schéma jako Python třídu a validujete DataFrame jedním dekorátorem. Verze 0.20 (2025) přidala plnou podporu Polars a 0.22 z roku 2026 vylepšila integraci s Pydantic.

import pandera as pa
from pandera.typing import DataFrame, Series

class OrdersSchema(pa.DataFrameModel):
    order_id: Series[int] = pa.Field(unique=True, gt=0)
    customer_id: Series[int] = pa.Field(nullable=False)
    amount_usd: Series[float] = pa.Field(ge=0, le=100_000)
    created_at: Series[pd.Timestamp]

    class Config:
        strict = True  # extra sloupce vyhazuji chybu

@pa.check_types
def aggregate_revenue(orders: DataFrame[OrdersSchema]) -> pd.DataFrame:
    return orders.groupby("customer_id")["amount_usd"].sum().reset_index()

Co tohle dělá: dekorátor @pa.check_types validuje vstupní DataFrame před spuštěním funkce. Když přijde špatný typ nebo null v customer_id, dostanete SchemaError s přesnou pozicí problému. Tohle už několikrát zachránilo můj tým před backfillem celého týdne. Chyba se odhalila u prvního řádku, ne až po půl hodině transformace. V minulém projektu mi to ušetřilo asi šest hodin debuggingu.

dbt tests a generic tests pro warehouse vrstvu

Když pipeline končí v data warehouse (Snowflake, BigQuery, Postgres), dbt je standard. dbt 1.8 (vydáno 2024) zavedlo unit tests pro SQL modely, což byla dlouho zoufale chybějící součást. Pro datovou kvalitu používám čtyři typy testů:

• Generic tests: unique, not_null, accepted_values, relationships, definované v YAML, běží na každém modelu.
• Singular tests: vlastní SQL dotaz, který vrací řádky, jež nemají existovat (pokud vrátí cokoliv, test selhal).
• Unit tests: mock vstupů jako CSV bloky v YAML, validace transformace bez warehouse.
• Data contracts: garance schématu, sloupce a typy se definují v YAML a porušení blokuje build.

# models/marts/orders.yml
version: 2

models:
  - name: fct_orders
    config:
      contract:
        enforced: true
    columns:
      - name: order_id
        data_type: bigint
        tests:
          - unique
          - not_null
      - name: customer_id
        data_type: bigint
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: amount_usd
        data_type: numeric
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= 0"

unit_tests:
  - name: test_revenue_aggregation
    model: fct_orders
    given:
      - input: ref('stg_orders')
        rows:
          - {order_id: 1, customer_id: 10, amount_cents: 1500}
          - {order_id: 2, customer_id: 10, amount_cents: 2300}
    expect:
      rows:
        - {order_id: 1, customer_id: 10, amount_usd: 15.00}
        - {order_id: 2, customer_id: 10, amount_usd: 23.00}

Idempotence, backfill a testování času

Tohle je sekce, kterou většina blogů přeskočí, ale pro mě je nejdůležitější. Pipeline musí být idempotentní. Spuštění té samé úlohy pro stejné datum dvakrát musí dát identický výsledek. Když idempotenci nemáte, backfill (znovuspuštění pipeline za poslední týden po výpadku) zduplikuje data nebo je nedeterministicky promíchá. Tohle jsem si naposledy odnesla v lednu, kdy nám výpadek warehouse zhltl tři dny dat a backfill je rozsypal kvůli chybějícímu MERGE.

Testovat to lze takhle:

from datetime import date
import pandas as pd

def test_pipeline_is_idempotent(tmp_path, raw_data):
    target = tmp_path / "warehouse.parquet"

    # Prvni beh
    run_pipeline(raw_data, output=target, run_date=date(2026, 6, 22))
    first = pd.read_parquet(target)

    # Druhy beh stejneho data, musi dat identicky vysledek
    run_pipeline(raw_data, output=target, run_date=date(2026, 6, 22))
    second = pd.read_parquet(target)

    pd.testing.assert_frame_equal(first, second)

def test_no_data_leak_across_partitions(raw_data):
    """Run za 22.6. nesmi ovlivnit data 21.6."""
    result_22 = run_pipeline(raw_data, run_date=date(2026, 6, 22))
    result_21 = run_pipeline(raw_data, run_date=date(2026, 6, 21))

    assert result_22["partition_date"].nunique() == 1
    assert result_21["partition_date"].nunique() == 1

Druhá věc, na kterou narazíte: testování kódu, který závisí na datetime.now(). Používejte freezegun a injektujte „aktuální" čas jako parametr. Funkce, která si sama volá now(), se prakticky testovat nedá. Pokud děláte pokročilejší práci s časovými řadami, koukněte i na můj průvodce analýzou časových řad v Pandas, kde rozebíráme práci s časovými indexy.

Integrace do CI/CD pipeline

Testy, které neběhají automaticky, jsou prakticky bezcenné. Tady je GitHub Actions konfigurace, kterou používám:

# .github/workflows/data-quality.yml
name: Data pipeline tests

on:
  pull_request:
    paths:
      - 'pipelines/**'
      - 'models/**'
      - 'tests/**'

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: test
        options: --health-cmd pg_isready

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.13'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pandera great-expectations dbt-postgres

      - name: Run pytest
        run: pytest tests/ -v --cov=pipelines --cov-fail-under=80

      - name: Run dbt tests
        run: |
          dbt deps
          dbt seed --target ci
          dbt run --target ci
          dbt test --target ci

Nejčastější chyby při testování datových pipeline

Z mojí zkušenosti se opakují tyhle vzorce:

• Testování proti produkčním datům. Lákavé, ale data se mění a test, který v pondělí prošel, ve středu spadne bez změny kódu. Vždycky fixture data v kódu nebo v seeds/.
• Žádný test na schéma. Když upstream přidá sloupec, většina kódu to projde, ale agregace dají jiný výsledek. Pandera strict=True nebo dbt contract vás zachrání.
• Testy, které mají side effecty. Pipeline test, který zapíše do produkční tabulky, je jen otázkou času, kdy způsobí incident. Vždycky tmp_path nebo testovací schéma.
• Validace až po zápisu. Pokud Great Expectations spustíte až po tom, co data zapíšete do warehouse, znečištění už proběhlo. Validujte před zápisem.
• Neexistující test na duplicity. ExpectColumnValuesToBeUnique nebo dbt unique test: pokud ho na primárním klíči nemáte, nikdy se nedozvíte, že vám pipeline duplikuje řádky.

Pokud na pipeline navazujete machine learning modely, doporučuju si přečíst i můj nedávný průvodce nasazením ML modelů s FastAPI. Tam navazujeme tématem testování modelů, což je další samostatná disciplína. A pokud řešíte čištění dat před vstupem do pipeline, můj průvodce čištěním dat v Pandas jde do hloubky preprocessing kroků.

Často kladené otázky

Jak testovat datovou pipeline v Pythonu?

Použijte dvě vrstvy: pytest pro logiku transformací (čisté funkce, malé fixture DataFrames) a Great Expectations nebo Pandera pro validaci reálných dat tekoucích pipeline. Pytest běží v CI při každém commitu, kvalitativní kontroly běží přímo v pipeline a zastaví ji, pokud data nesplňují očekávání.

Co je Great Expectations a kdy ho použít?

Great Expectations je framework pro deklarativní popis očekávání o datech (např. „sloupec order_id musí být unikátní a non-null"). Použijte ho, pokud potřebujete bohatý reporting, integraci s Airflow/Dagster a sdílení expectací mezi týmy. Pro jednoduší case je Pandera lehčí volba.

Jaký je rozdíl mezi Pandera a Great Expectations?

Pandera je odlehčená knihovna integrovaná přímo do Python typování. Schéma definujete jako třídu a validace běží přes dekorátor. Great Expectations je framework s vlastním stavem, store a reportingem. Pandera je lepší pro in-process validaci, Great Expectations pro cross-team data quality monitoring.

Jak otestovat idempotenci datové pipeline?

Spusťte pipeline dvakrát se stejným vstupem a stejným run_date, porovnejte výsledek pomocí pd.testing.assert_frame_equal. Pokud se výsledky liší, pipeline není idempotentní a backfill ji rozbije. Klíčové je, aby zápis používal MERGE nebo INSERT … ON CONFLICT, ne prostý INSERT.

Mají se data quality testy spouštět před nebo po zápisu do warehouse?

Vždy před zápisem do produkční tabulky. Pokud validujete až po, znečištění už proběhlo a budete dělat bolestivý rollback nebo backfill. Vzor: načti, transformuj, validuj (Pandera/GX), zapiš do staging, další test (dbt), swap se produkční tabulkou.

Stačí dbt tests, nebo potřebuju i pytest?

Stačí jen pokud máte čistě SQL pipeline. Jakmile máte Python transformace (úprava JSON, ML feature engineering, custom parsing), dbt tests Python kód netestují a potřebujete pytest. V praxi mívám obě sady: pytest pro Python kód, dbt tests pro SQL modely v warehouse.