Von der CSV zu Erkenntnissen: ein praxisnahes Playbook

import pandas as pd

df = pd.read_csv("data/dog_breeds.csv")

# Quick sanity pass
print(df.shape)               # (628, 44)
print(df["breed_group"].value_counts().head())
print(df["origin"].nunique(), "unique origins")

# Derived lifespan midpoint
df["lifespan_mid"] = (df["min_life_expectancy"] + df["max_life_expectancy"]) / 2
top = df.nlargest(10, "popularity_score")[["name", "breed_group", "origin"]]

import polars as pl

df = pl.read_csv("data/dog_breeds.csv")
big_hounds = (
    df.filter((pl.col("breed_group") == "Hound") & (pl.col("size_category") == "large"))
      .select(["name", "origin", "max_weight_male"])
      .sort("max_weight_male", descending=True)
)

-- Count sources per breed
SELECT b.name, COUNT(s.source_url) AS source_count
FROM read_csv_auto('data/dog_breeds.csv') b
LEFT JOIN read_csv_auto('data/dog_breed_sources.csv') s
  ON s.breed_key = b.key
GROUP BY b.name
ORDER BY source_count DESC
LIMIT 10;

-- Find breeds missing primary images
SELECT b.key, b.name
FROM read_csv_auto('data/dog_breeds.csv') b
LEFT JOIN (
  SELECT breed_key FROM read_csv_auto('data/dog_breed_images_inventory.csv')
  WHERE is_primary = 1
) i ON i.breed_key = b.key
WHERE i.breed_key IS NULL;

import sqlite3, pandas as pd

con = sqlite3.connect("data/dog_breeds.sqlite")

# Widest lifespan ranges — good candidates for outlier review
q = """
SELECT name, min_life_expectancy, max_life_expectancy,
       (max_life_expectancy - min_life_expectancy) AS span
FROM dogs
WHERE min_life_expectancy IS NOT NULL
ORDER BY span DESC
LIMIT 20
"""
outliers = pd.read_sql_query(q, con)

import great_expectations as gx

ctx = gx.get_context()
src = ctx.sources.add_pandas("dogs")
asset = src.add_csv_asset("dog_breeds", filepath_or_buffer="data/dog_breeds.csv")

batch = asset.build_batch_request()
validator = ctx.get_validator(batch_request=batch)

validator.expect_column_values_to_not_be_null("key")
validator.expect_column_values_to_be_unique("key")
validator.expect_column_values_to_be_between("energy", 1, 5)
validator.expect_column_values_to_be_between("trainability", 1, 5)
validator.expect_column_values_to_match_regex("source_url", r"^https?://")
validator.expect_column_values_to_be_in_set(
    "review_status", ["reviewed", "seeded"]
)
validator.save_expectation_suite("dog_breeds.basic")

from datetime import datetime
from airflow.decorators import dag, task

@dag(
    dag_id="dog_breeds_daily",
    start_date=datetime(2026, 4, 17),
    schedule="@daily",
    catchup=False,
    tags=["dogs", "demo"],
)
def dog_breeds_pipeline():

    @task
    def ingest() -> str:
        import shutil
        src = "/opt/airflow/data/dog_breeds.csv"
        dst = "/opt/airflow/warehouse/raw/dog_breeds.csv"
        shutil.copyfile(src, dst)
        return dst

    @task
    def validate(path: str) -> str:
        import pandas as pd
        df = pd.read_csv(path)
        assert df["key"].is_unique, "breed key must be unique"
        assert df[["energy", "trainability"]].max().max() <= 5
        return path

    @task
    def transform(path: str) -> str:
        import duckdb
        duckdb.sql(f"""
            CREATE OR REPLACE TABLE mart_breed_profile AS
            SELECT key, name, breed_group, origin, size_category,
                   (min_life_expectancy + max_life_expectancy) / 2.0 AS lifespan_mid,
                   popularity_score
            FROM read_csv_auto('{path}')
        """)
        duckdb.sql("EXPORT DATABASE '/opt/airflow/warehouse/marts' (FORMAT PARQUET)")
        return "/opt/airflow/warehouse/marts"

    @task
    def publish(_marts: str):
        # push to S3, Snowflake, BigQuery, ...
        print("published ✅")

    publish(transform(validate(ingest())))

dog_breeds_pipeline()

-- models/staging/stg_breeds.sql
select
    key                                    as breed_key,
    name                                   as breed_name,
    nullif(breed_group, '')                as breed_group,
    nullif(origin, '')                     as origin,
    lower(size_category)                   as size_category,
    (min_life_expectancy + max_life_expectancy) / 2.0 as lifespan_mid_years,
    popularity_score
from {{ source('raw', 'dog_breeds') }};

-- models/marts/dim_breed.sql
select
    b.*,
    s.source_registry,
    s.source_url
from {{ ref('stg_breeds') }} b
left join {{ source('raw', 'dog_breed_sources') }} s
       on s.breed_key = b.breed_key and s.is_primary = 1;