Veri Pipeline
ETL, Lakehouse & Polars.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.operators.s3 import S3DeleteObjectsOperator
from airflow.utils.dates import days_ago
from datetime import timedelta
import polars as pl

DEFAULT_ARGS = {
    "owner": "data-eng",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["alerts@company.com"],
}

def extract(**context):
    # S3'ten parquet oku
    ds = context["ds"]  # YYYY-MM-DD
    df = pl.read_parquet(f"s3://raw-bucket/events/date={ds}/*.parquet")
    context["ti"].xcom_push(key="row_count", value=len(df))
    df.write_parquet(f"/tmp/extract_{ds}.parquet")

def transform(**context):
    ds = context["ds"]
    df = pl.read_parquet(f"/tmp/extract_{ds}.parquet")
    df = (
        df
        .filter(pl.col("event_type").is_in(["click", "purchase", "view"]))
        .with_columns([
            pl.col("ts").cast(pl.Datetime).alias("event_time"),
            (pl.col("revenue").fill_null(0.0)).alias("revenue"),
        ])
        .drop_nulls(subset=["user_id", "event_type"])
    )
    df.write_parquet(f"/tmp/transform_{ds}.parquet")

def load_to_delta(**context):
    from deltalake import write_deltalake
    ds = context["ds"]
    df = pl.read_parquet(f"/tmp/transform_{ds}.parquet")
    write_deltalake(
        "s3://curated-bucket/events",
        df.to_arrow(),
        mode="append",
        partition_by=["event_type"],
    )

with DAG(
    dag_id="daily_events_pipeline",
    default_args=DEFAULT_ARGS,
    schedule_interval="@daily",
    start_date=days_ago(1),
    catchup=False,
    tags=["events", "production"],
) as dag:

    t_extract   = PythonOperator(task_id="extract",    python_callable=extract,        provide_context=True)
    t_transform = PythonOperator(task_id="transform",   python_callable=transform,      provide_context=True)
    t_load      = PythonOperator(task_id="load_delta",  python_callable=load_to_delta,  provide_context=True)

    t_extract >> t_transform >> t_load

from prefect import flow, task
from prefect.tasks import exponential_backoff
import polars as pl

@task(retries=3, retry_delay_seconds=exponential_backoff(backoff_factor=2))
def extract_events(date: str) -> pl.DataFrame:
    return pl.read_parquet(f"s3://raw-bucket/events/date={date}/*.parquet")

@task
def transform_events(df: pl.DataFrame) -> pl.DataFrame:
    return (
        df.filter(pl.col("event_type") != "bot")
          .with_columns(pl.col("ts").cast(pl.Datetime))
    )

@task
def load_events(df: pl.DataFrame, date: str):
    from deltalake import write_deltalake
    write_deltalake("s3://curated-bucket/events", df.to_arrow(), mode="append")

@flow(name="events-pipeline", log_prints=True)
def events_pipeline(date: str = "2026-01-01"):
    raw  = extract_events(date)
    clean = transform_events(raw)
    load_events(clean, date)
    print(f"Pipeline tamamlandı: {len(clean)} satır yüklendi.")

if __name__ == "__main__":
    events_pipeline(date="2026-04-12")

import pyarrow as pa
from deltalake import DeltaTable, write_deltalake

TABLE_PATH = "s3://my-bucket/silver/users"
STORAGE_OPTIONS = {"AWS_REGION": "us-east-1", "AWS_S3_ALLOW_UNSAFE_RENAME": "true"}

# ── İlk yazma ──────────────────────────────────────────────────────
schema = pa.schema([
    pa.field("user_id",   pa.int64()),
    pa.field("name",      pa.string()),
    pa.field("country",   pa.string()),
    pa.field("created_at", pa.timestamp("us")),
])
data = pa.table({
    "user_id":   [1, 2, 3],
    "name":      ["Alice", "Bob", "Carol"],
    "country":   ["TR", "DE", "US"],
    "created_at": pa.array([1_700_000_000_000_000] * 3, type=pa.timestamp("us")),
})
write_deltalake(TABLE_PATH, data, schema=schema,
               mode="overwrite", partition_by=["country"],
               storage_options=STORAGE_OPTIONS)

# ── UPSERT (MERGE) ──────────────────────────────────────────────────
dt = DeltaTable(TABLE_PATH, storage_options=STORAGE_OPTIONS)
new_data = pa.table({
    "user_id":   [2, 4],
    "name":      ["Bobby", "Dave"],
    "country":   ["DE", "FR"],
    "created_at": pa.array([1_710_000_000_000_000] * 2, type=pa.timestamp("us")),
})
(
    dt.merge(
        source=new_data,
        predicate="s.user_id = t.user_id",
        source_alias="s",
        target_alias="t",
    )
    .when_matched_update_all()
    .when_not_matched_insert_all()
    .execute()
)

# ── Time travel ─────────────────────────────────────────────────────
dt_v0 = DeltaTable(TABLE_PATH, version=0, storage_options=STORAGE_OPTIONS)
print("Version 0:", dt_v0.to_pandas())

import polars as pl
from datetime import date

# ── Lazy pipeline (scan, filter, aggregate) ────────────────────────
events = (
    pl.scan_parquet("s3://bucket/events/**/*.parquet")
    .filter(
        (pl.col("event_date") >= date(2026, 1, 1)) &
        (pl.col("event_type").is_in(["purchase", "add_to_cart"]))
    )
    .with_columns([
        (pl.col("revenue") * pl.col("quantity")).alias("total_value"),
        pl.col("user_id").cast(pl.Utf8).alias("user_id_str"),
    ])
)

# ── Aggregation ────────────────────────────────────────────────────
user_stats = (
    events
    .group_by(["user_id", "event_date"])
    .agg([
        pl.col("total_value").sum().alias("daily_revenue"),
        pl.col("event_type").count().alias("event_count"),
        pl.col("product_id").n_unique().alias("unique_products"),
    ])
    .sort(["user_id", "event_date"])
)

# ── Join ile kullanıcı profili ─────────────────────────────────────
users = pl.scan_parquet("s3://bucket/users/*.parquet").select(["user_id", "country", "tier"])

result = (
    user_stats
    .join(users, on="user_id", how="left")
    .filter(pl.col("tier") == "premium")
    .collect(streaming=True)  # büyük dataset için streaming mode
)
print(result.head())

# ── Window function (rolling 7 gün ortalama) ───────────────────────
rolling = (
    result
    .sort(["user_id", "event_date"])
    .with_columns(
        pl.col("daily_revenue")
          .rolling_mean(window_size=7)
          .over("user_id")
          .alias("rev_7d_avg")
    )
)

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = (
    SparkSession.builder
    .appName("UserFeaturePipeline")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .config("spark.sql.adaptive.enabled", "true")   # AQE — Adaptive Query Execution
    .config("spark.sql.shuffle.partitions", "200")
    .getOrCreate()
)

# ── Okuma ─────────────────────────────────────────────────────────
events = spark.read.format("delta").load("s3://bucket/silver/events")
users  = spark.read.format("parquet").load("s3://bucket/bronze/users")

# ── Transformations (lazy) ─────────────────────────────────────────
enriched = (
    events
    .filter(F.col("event_type") == "purchase")
    .withColumn("revenue_usd", F.col("revenue") * F.col("fx_rate"))
    .join(users.select("user_id", "country", "tier"), on="user_id", how="left")
    .groupBy("user_id", "country")
    .agg(
        F.sum("revenue_usd").alias("total_revenue"),
        F.count("*").alias("purchase_count"),
        F.avg("revenue_usd").alias("avg_order_value"),
    )
)

# ── Action — Delta'ya yaz ─────────────────────────────────────────
(
    enriched
    .repartition(50, "country")   # partition pruning için
    .write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .partitionBy("country")
    .save("s3://bucket/gold/user_features")
)

spark.stop()

import great_expectations as gx
import pandas as pd

context = gx.get_context()

# ── Datasource tanımla ──────────────────────────────────────────────
datasource = context.sources.add_or_update_pandas(name="events_ds")
asset = datasource.add_dataframe_asset(name="events")
batch_request = asset.build_batch_request(dataframe=pd.read_parquet("events.parquet"))

# ── Expectation Suite ───────────────────────────────────────────────
suite = context.add_or_update_expectation_suite("events_suite")
validator = context.get_validator(batch_request=batch_request, expectation_suite=suite)

validator.expect_column_to_exist("user_id")
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_between("revenue", min_value=0, max_value=100_000)
validator.expect_column_values_to_be_in_set("event_type", ["click", "purchase", "view"])
validator.expect_column_pair_values_to_be_equal("order_id", "order_id_copy")
validator.expect_table_row_count_to_be_between(min_value=1000, max_value=10_000_000)

validator.save_expectation_suite()

# ── Checkpoint çalıştır ─────────────────────────────────────────────
checkpoint = context.add_or_update_checkpoint(
    name="events_checkpoint",
    validator=validator,
)
result = checkpoint.run()
if not result["success"]:
    raise ValueError("Data quality check başarısız! Pipeline durduruluyor.")

import polars as pl
import pytest
from pipeline.transforms import clean_events, aggregate_daily

@pytest.fixture
def raw_events():
    return pl.DataFrame({
        "user_id":    [1, 2, 2, 3, None],
        "event_type": ["click", "purchase", "bot", "purchase", "view"],
        "revenue":    [0.0, 49.99, 0.0, 120.0, 0.0],
    })

def test_clean_removes_bot_events(raw_events):
    result = clean_events(raw_events)
    assert "bot" not in result["event_type"].to_list()

def test_clean_drops_null_user_id(raw_events):
    result = clean_events(raw_events)
    assert result["user_id"].null_count() == 0

def test_aggregate_daily_revenue(raw_events):
    clean = clean_events(raw_events)
    agg   = aggregate_daily(clean)
    user2_revenue = agg.filter(pl.col("user_id") == 2)["total_revenue"][0]
    assert user2_revenue == pytest.approx(49.99, rel=1e-3)

-- Z-ORDER ile optimize (Spark SQL / Databricks)
OPTIMIZE delta.`s3://bucket/gold/user_features`
ZORDER BY (user_id, event_date);

-- 7 günden eski versiyon dosyalarını temizle
VACUUM delta.`s3://bucket/gold/user_features` RETAIN 168 HOURS;

-- Partition statistics güncelle
ANALYZE TABLE user_features COMPUTE STATISTICS FOR ALL COLUMNS;

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
from datetime import timedelta

@dag(
    dag_id="s3_to_feature_table",
    schedule_interval="0 3 * * *",   # her gün 03:00
    start_date=days_ago(1),
    default_args={"retries": 2, "retry_delay": timedelta(minutes=10)},
    catchup=False,
)
def pipeline():

    @task
    def extract(ds=None):
        import polars as pl
        df = pl.scan_parquet(f"s3://raw/events/date={ds}/*.parquet").collect()
        assert len(df) > 0, "Boş dataset!"
        return df.write_ipc(None)  # Arrow IPC bytes — XCom için serialize

    @task
    def validate(ipc_bytes: bytes):
        import polars as pl, io
        df = pl.read_ipc(io.BytesIO(ipc_bytes))
        required_cols = {"user_id", "event_type", "revenue", "ts"}
        assert required_cols.issubset(set(df.columns)), "Eksik kolon!"
        assert df["revenue"].min() >= 0, "Negatif revenue!"
        return ipc_bytes

    @task
    def write_silver(ipc_bytes: bytes, ds=None):
        import polars as pl, io
        from deltalake import write_deltalake
        df = pl.read_ipc(io.BytesIO(ipc_bytes))
        write_deltalake("s3://curated/silver/events", df.to_arrow(),
                        mode="append", partition_by=["event_date"])

    @task
    def build_features(ds=None):
        from pyspark.sql import SparkSession
        import pyspark.sql.functions as F
        spark = SparkSession.builder.getOrCreate()
        df = spark.read.format("delta").load("s3://curated/silver/events")
        features = (
            df.groupBy("user_id")
              .agg(
                F.sum("revenue").alias("ltv"),
                F.count("*").alias("total_events"),
                F.countDistinct("event_date").alias("active_days"),
              )
        )
        features.write.format("delta").mode("overwrite").save("s3://curated/gold/user_features")
        spark.stop()

    raw = extract()
    validated = validate(raw)
    write_silver(validated) >> build_features()

pipeline()