Workshop Setup — Postgres, MinIO, Kafka, Spark, Jupyter with uv + Docker Compose

# Beklenen çıktı (özetlenmiş):
$ uv add jupyter pandas polars duckdb pyarrow ...
Resolved 142 packages in 1.34s
Prepared 142 packages in 4.12s
Installed 142 packages in 187ms
 + duckdb==1.1.3
 + jupyter==1.1.1
 + kafka-python==2.0.2
 + pandas==2.2.3
 + polars==1.18.0
 + psycopg2-binary==2.9.10
 + pyarrow==18.1.0
 + pyspark==3.5.3
 + ...

# docker-compose.yml
name: ai-veri-muh
 
services:
  # ============================================================
  # Postgres 16 + pgvector — OLTP + vector store
  # ============================================================
  postgres:
    image: pgvector/pgvector:pg16
    container_name: ai-veri-postgres
    restart: unless-stopped
    environment:
      POSTGRES_USER: dataeng
      POSTGRES_PASSWORD: dataeng_pw_2026
      POSTGRES_DB: lab
      POSTGRES_INITDB_ARGS: "--encoding=UTF-8 --locale=C"
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./docker/postgres/init.sql:/docker-entrypoint-initdb.d/init.sql:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U dataeng -d lab"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - lab-net
 
  # ============================================================
  # MinIO — S3-uyumlu object store
  # ============================================================
  minio:
    image: minio/minio:latest
    container_name: ai-veri-minio
    restart: unless-stopped
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin_pw_2026
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"   # API
      - "9001:9001"   # Console UI
    volumes:
      - minio_data:/data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 10s
      retries: 5
    networks:
      - lab-net
 
  # MinIO bucket'larını hazırlamak için one-shot init container
  minio-init:
    image: minio/mc:latest
    container_name: ai-veri-minio-init
    depends_on:
      minio:
        condition: service_healthy
    entrypoint: >
      /bin/sh -c "
      mc alias set local http://minio:9000 minioadmin minioadmin_pw_2026;
      mc mb -p local/raw 2>/dev/null || true;
      mc mb -p local/lakehouse 2>/dev/null || true;
      mc mb -p local/landing 2>/dev/null || true;
      mc anonymous set public local/raw;
      echo '✅ MinIO bucket setup complete';
      exit 0;
      "
    networks:
      - lab-net
 
  # ============================================================
  # Kafka (KRaft mode — Zookeeper yok!)
  # ============================================================
  kafka:
    image: apache/kafka:3.8.0
    container_name: ai-veri-kafka
    restart: unless-stopped
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"
    volumes:
      - kafka_data:/var/lib/kafka/data
    networks:
      - lab-net
 
  # ============================================================
  # Schema Registry — Avro/Protobuf schema yönetimi
  # ============================================================
  schema-registry:
    image: confluentinc/cp-schema-registry:7.6.0
    container_name: ai-veri-schema-registry
    restart: unless-stopped
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
    networks:
      - lab-net
 
  # ============================================================
  # Spark master + worker
  # ============================================================
  spark-master:
    image: bitnami/spark:3.5.3
    container_name: ai-veri-spark-master
    restart: unless-stopped
    environment:
      SPARK_MODE: master
      SPARK_RPC_AUTHENTICATION_ENABLED: "no"
      SPARK_RPC_ENCRYPTION_ENABLED: "no"
      SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED: "no"
      SPARK_SSL_ENABLED: "no"
    ports:
      - "7077:7077"   # master
      - "8080:8080"   # UI
    networks:
      - lab-net
 
  spark-worker:
    image: bitnami/spark:3.5.3
    container_name: ai-veri-spark-worker
    restart: unless-stopped
    depends_on:
      - spark-master
    environment:
      SPARK_MODE: worker
      SPARK_MASTER_URL: spark://spark-master:7077
      SPARK_WORKER_MEMORY: 2G
      SPARK_WORKER_CORES: 2
      SPARK_RPC_AUTHENTICATION_ENABLED: "no"
      SPARK_RPC_ENCRYPTION_ENABLED: "no"
      SPARK_SSL_ENABLED: "no"
    networks:
      - lab-net
 
  # ============================================================
  # JupyterLab — interactive notebook environment
  # ============================================================
  jupyter:
    image: jupyter/pyspark-notebook:python-3.11
    container_name: ai-veri-jupyter
    restart: unless-stopped
    user: root
    environment:
      JUPYTER_TOKEN: "dataeng2026"
      GRANT_SUDO: "yes"
      JUPYTER_ENABLE_LAB: "yes"
      SPARK_OPTS: "--master=spark://spark-master:7077"
    ports:
      - "8888:8888"
    volumes:
      - ./notebooks:/home/jovyan/work
      - jupyter_data:/home/jovyan/.local
    networks:
      - lab-net
 
volumes:
  postgres_data:
  minio_data:
  kafka_data:
  jupyter_data:
 
networks:
  lab-net:
    driver: bridge
 

-- docker/postgres/init.sql
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
 
-- Lab şeması
CREATE SCHEMA IF NOT EXISTS lab;
GRANT ALL PRIVILEGES ON SCHEMA lab TO dataeng;
 
-- "Hello pipeline" için örnek tablo
CREATE TABLE IF NOT EXISTS lab.events (
  event_id      BIGSERIAL PRIMARY KEY,
  event_type    TEXT NOT NULL,
  payload       JSONB NOT NULL,
  embedding     VECTOR(384),  -- pgvector!
  created_at    TIMESTAMPTZ DEFAULT now()
);
 
CREATE INDEX idx_events_type ON lab.events(event_type);
CREATE INDEX idx_events_created ON lab.events(created_at DESC);
 
COMMENT ON TABLE lab.events IS 'AI veri mühendisliği kursu — örnek event tablosu';

# === HELLO PIPELINE — Lokal stack entegrasyon testi ===
 
# 1) Bağımlılıklar (notebook içinde olduğun için tek seferlik install)
import subprocess
subprocess.run(["pip", "install", "-q", "psycopg2-binary", "minio", "kafka-python", "duckdb", "pyarrow", "pandas"], check=True)
 
import psycopg2, json, random, uuid, io
from datetime import datetime
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from minio import Minio
from kafka import KafkaProducer
import duckdb
 
# 2) Veriyi üret — 10K sahte event
print("🎲 Sahte event'ler üretiliyor...")
events = [
    {
        "event_id": str(uuid.uuid4()),
        "event_type": random.choice(["page_view", "click", "purchase", "search"]),
        "user_id": f"user_{random.randint(1, 1000)}",
        "amount": round(random.uniform(0, 500), 2),
        "ts": datetime.utcnow().isoformat(),
    }
    for _ in range(10_000)
]
df = pd.DataFrame(events)
print(f"   ✅ {len(df):,} event üretildi")
 
# 3) Postgres'e yaz
print("\n🐘 Postgres'e yazılıyor...")
conn = psycopg2.connect(
    host="postgres", dbname="lab", user="dataeng", password="dataeng_pw_2026"
)
cur = conn.cursor()
# İlk 100'ünü Postgres'e
for ev in events[:100]:
    cur.execute(
        "INSERT INTO lab.events (event_type, payload) VALUES (%s, %s)",
        (ev["event_type"], json.dumps(ev)),
    )
conn.commit()
cur.execute("SELECT COUNT(*) FROM lab.events;")
pg_count = cur.fetchone()[0]
print(f"   ✅ Postgres'te şu an {pg_count} satır var")
cur.close()
conn.close()
 
# 4) MinIO'ya Parquet olarak yaz
print("\n📦 MinIO'ya Parquet yazılıyor...")
mc = Minio(
    "minio:9000",
    access_key="minioadmin",
    secret_key="minioadmin_pw_2026",
    secure=False,
)
table = pa.Table.from_pandas(df)
buf = io.BytesIO()
pq.write_table(table, buf, compression="snappy")
buf.seek(0)
mc.put_object(
    "lakehouse",
    "hello/events.parquet",
    buf,
    length=buf.getbuffer().nbytes,
    content_type="application/octet-stream",
)
print(f"   ✅ s3://lakehouse/hello/events.parquet yazıldı ({buf.getbuffer().nbytes / 1024:.1f} KB)")
 
# 5) Kafka'ya event publish
print("\n📡 Kafka'ya yayın yapılıyor...")
producer = KafkaProducer(
    bootstrap_servers="kafka:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)
for ev in events[:50]:
    producer.send("hello-topic", ev)
producer.flush()
print("   ✅ 50 event 'hello-topic' topic'ine gönderildi")
 
# 6) DuckDB ile MinIO'daki Parquet'i sorgula (no-download!)
print("\n🦆 DuckDB ile MinIO'daki Parquet'i sorgula...")
conn = duckdb.connect()
conn.execute("INSTALL httpfs; LOAD httpfs;")
conn.execute("""
    SET s3_endpoint='minio:9000';
    SET s3_access_key_id='minioadmin';
    SET s3_secret_access_key='minioadmin_pw_2026';
    SET s3_use_ssl=false;
    SET s3_url_style='path';
""")
result = conn.execute("""
    SELECT event_type, COUNT(*) AS n, ROUND(SUM(amount), 2) AS total
    FROM read_parquet('s3://lakehouse/hello/events.parquet')
    GROUP BY 1
    ORDER BY n DESC;
""").df()
print(result.to_string(index=False))
 
print("\n🎉 HELLO PIPELINE BAŞARILI — bütün servisler birlikte çalışıyor!")