atc-agents/api/hadoop_analytics.py

"""Hadoop lakehouse analytics for the Command Center 'Hadoop' tab.

- /api/hadoop/filters    -> available filter values (years/regions/categories/channels)
- /api/hadoop/analytics  -> live, filterable KPIs + breakdowns via Trino (with exec stats)
- /api/hadoop/source     -> data lineage: exactly where each engine reads from
- /api/hadoop/engines    -> engine comparison. Trino is measured live; Hive is a REAL
                            measurement (Apache Hive 3.1.3, MapReduce) refreshed over SSH
                            from the cluster; Impala is a clearly-labelled representative.

Trino reads the curated table iceberg.hadoop.historical_sales (Parquet on Dell ECS S3);
Hive reads lake.historical_sales (CSV on HDFS). Both ~25k rows -> fair latency contrast.
"""

from __future__ import annotations

import asyncio
import json
import os
import time
from pathlib import Path
from typing import Any

import httpx
from fastapi import APIRouter, Query
from fastapi.responses import JSONResponse

TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/")
TRINO_USER = os.getenv("TRINO_USER", "mo")
TABLE = "iceberg.hadoop.historical_sales"
S3_ENDPOINT = os.getenv("S3_ENDPOINT", "http://10.0.20.111:9020")
HDFS_NN_URL = os.getenv("HDFS_NN_URL", "http://10.0.21.61:9870")

HIVE_SSH_HOST = os.getenv("HIVE_SSH_HOST", "10.0.21.61")
HIVE_SSH_USER = os.getenv("HIVE_SSH_USER", "root")
HIVE_BENCH_CMD = os.getenv("HIVE_BENCH_CMD", "/opt/hive/bench.sh")
HIVE_BENCH_TTL = float(os.getenv("HIVE_BENCH_TTL", "1800"))  # 30 min
_SEED = Path(__file__).with_name("hive_bench_seed.json")
_PERSIST = Path("/data/hive_bench.json")

router = APIRouter(prefix="/api/hadoop", tags=["hadoop"])

_cache: dict[str, Any] = {}
_TTL = 45.0

# dimension -> Iceberg column
DIMS = {"years": "order_year", "regions": "region", "categories": "product_category", "channels": "channel"}


# --------------------------------------------------------------------------- Trino
async def _trino(sql: str, deadline_s: float = 30.0) -> dict[str, Any]:
    headers = {"X-Trino-User": TRINO_USER, "X-Trino-Catalog": "iceberg", "X-Trino-Schema": "hadoop"}
    rows: list[list[Any]] = []
    cols: list[str] | None = None
    stats: dict[str, Any] = {}
    start = time.monotonic()
    async with httpx.AsyncClient(timeout=15.0) as client:
        r = await client.post(f"{TRINO_URL}/v1/statement", content=sql.encode(), headers=headers)
        r.raise_for_status()
        payload = r.json()
        while True:
            if payload.get("error"):
                raise RuntimeError(payload["error"].get("message", str(payload["error"])))
            if payload.get("columns") and cols is None:
                cols = [c["name"] for c in payload["columns"]]
            rows.extend(payload.get("data", []) or [])
            if payload.get("stats"):
                stats = payload["stats"]
            nxt = payload.get("nextUri")
            if not nxt:
                break
            if time.monotonic() - start > deadline_s:
                raise TimeoutError("Trino query exceeded deadline")
            await asyncio.sleep(0.04)
            rr = await client.get(nxt, headers={"X-Trino-User": TRINO_USER})
            rr.raise_for_status()
            payload = rr.json()
    return {"rows": rows, "columns": cols or [], "stats": stats}


def _trino_stats(stats: dict[str, Any]) -> dict[str, Any]:
    return {
        "processed_rows": stats.get("processedRows"),
        "processed_bytes": stats.get("processedBytes"),
        "elapsed_ms": stats.get("elapsedTimeMillis"),
        "cpu_ms": stats.get("cpuTimeMillis"),
        "peak_memory_bytes": stats.get("peakMemoryBytes"),
        "splits": stats.get("totalSplits"),
    }


# --------------------------------------------------------------------------- filters
def _parse(v: str | None) -> list[str]:
    return [x.strip() for x in (v or "").split(",") if x.strip()]


def _where(years, regions, categories, channels) -> str:
    clauses = []
    yrs = [y for y in years if y.lstrip("-").isdigit()]
    if yrs:
        clauses.append(f"order_year IN ({','.join(yrs)})")
    for vals, col in ((regions, "region"), (categories, "product_category"), (channels, "channel")):
        if vals:
            esc = ",".join("'" + v.replace("'", "''") + "'" for v in vals)
            clauses.append(f"{col} IN ({esc})")
    return ("WHERE " + " AND ".join(clauses)) if clauses else ""


def _cached(key: str, ttl: float = _TTL):
    item = _cache.get(key)
    if item and (time.time() - item["ts"] < ttl):
        return item["data"]
    return None


def _store(key: str, data: Any):
    _cache[key] = {"data": data, "ts": time.time()}
    return data


@router.get("/filters")
async def filters():
    cached = _cached("filters", ttl=300)
    if cached is not None:
        return JSONResponse(cached)
    try:
        out = {}
        q = await asyncio.gather(
            _trino(f"SELECT DISTINCT order_year FROM {TABLE} ORDER BY 1"),
            _trino(f"SELECT DISTINCT region FROM {TABLE} ORDER BY 1"),
            _trino(f"SELECT DISTINCT product_category FROM {TABLE} ORDER BY 1"),
            _trino(f"SELECT DISTINCT channel FROM {TABLE} ORDER BY 1"),
        )
        out = {
            "ok": True,
            "years": [r[0] for r in q[0]["rows"]],
            "regions": [r[0] for r in q[1]["rows"]],
            "categories": [r[0] for r in q[2]["rows"]],
            "channels": [r[0] for r in q[3]["rows"]],
        }
        return JSONResponse(_store("filters", out))
    except Exception as e:
        return JSONResponse({"ok": False, "error": str(e)})


@router.get("/analytics")
async def analytics(
    years: str | None = Query(None),
    regions: str | None = Query(None),
    categories: str | None = Query(None),
    channels: str | None = Query(None),
):
    where = _where(_parse(years), _parse(regions), _parse(categories), _parse(channels))
    key = "an:" + where
    cached = _cached(key)
    if cached is not None:
        return JSONResponse(cached)
    try:
        kpis_q = (
            f"SELECT count(*), sum(amount), avg(amount), sum(quantity), count(DISTINCT region), "
            f"min(order_year), max(order_year), avg(unit_price) FROM {TABLE} {where}"
        )
        kpi_res = await _trino(kpis_q)
        k = kpi_res["rows"][0] if kpi_res["rows"] else [0] * 8
        by_year, by_region, by_category, by_channel, by_status = await asyncio.gather(
            _trino(f"SELECT order_year, sum(amount), count(*) FROM {TABLE} {where} GROUP BY order_year ORDER BY order_year"),
            _trino(f"SELECT region, sum(amount), count(*) FROM {TABLE} {where} GROUP BY region ORDER BY 2 DESC"),
            _trino(f"SELECT product_category, sum(amount), count(*) FROM {TABLE} {where} GROUP BY product_category ORDER BY 2 DESC"),
            _trino(f"SELECT channel, sum(amount), count(*) FROM {TABLE} {where} GROUP BY channel ORDER BY 2 DESC"),
            _trino(f"SELECT region, product_category, sum(amount) FROM {TABLE} {where} GROUP BY region, product_category ORDER BY 1,2"),
        )
        data = {
            "ok": True,
            "table": TABLE,
            "filtered": bool(where),
            "where": where,
            "kpis": {
                "orders": int(k[0] or 0),
                "revenue": float(k[1] or 0),
                "aov": float(k[2] or 0),
                "units": int(k[3] or 0),
                "regions": int(k[4] or 0),
                "year_min": k[5],
                "year_max": k[6],
                "avg_unit_price": float(k[7] or 0),
            },
            "by_year": [{"year": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_year["rows"]],
            "by_region": [{"region": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_region["rows"]],
            "by_category": [{"category": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_category["rows"]],
            "by_channel": [{"channel": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_channel["rows"]],
            "matrix": [{"region": r[0], "category": r[1], "revenue": float(r[2] or 0)} for r in by_status["rows"]],
            "trino_stats": _trino_stats(kpi_res["stats"]),
        }
        return JSONResponse(_store(key, data))
    except Exception as e:
        return JSONResponse({"ok": False, "error": str(e)})


@router.get("/source")
async def source():
    cached = _cached("source", ttl=120)
    if cached is not None:
        return JSONResponse(cached)
    trino_src: dict[str, Any] = {
        "engine": "Trino",
        "catalog": "iceberg",
        "schema": "hadoop",
        "table": "historical_sales",
        "format": "Apache Iceberg · Parquet",
        "storage": "Dell ECS S3 (object store)",
        "location": "s3://data/hadoop/historical_sales",
        "s3_endpoint": S3_ENDPOINT,
        "metastore": "Iceberg (file metadata on S3)",
    }
    try:
        f = await _trino('SELECT count(*), coalesce(sum(file_size_in_bytes),0) FROM iceberg.hadoop."historical_sales$files"')
        trino_src["data_files"] = int(f["rows"][0][0]) if f["rows"] else None
        trino_src["data_bytes"] = int(f["rows"][0][1]) if f["rows"] else None
        s = await _trino('SELECT snapshot_id, committed_at FROM iceberg.hadoop."historical_sales$snapshots" ORDER BY committed_at DESC LIMIT 1')
        if s["rows"]:
            trino_src["snapshot_id"] = str(s["rows"][0][0])
            trino_src["snapshot_at"] = str(s["rows"][0][1])
        c = await _trino(f"SELECT count(*) FROM {TABLE}")
        trino_src["rows"] = int(c["rows"][0][0]) if c["rows"] else None
    except Exception as e:
        trino_src["error"] = str(e)

    hv = _hive_cache["data"] or {}
    hive_src = {
        "engine": "Hive",
        "database": "lake",
        "table": "historical_sales",
        "format": hv.get("storage", "HDFS · CSV (TEXTFILE)"),
        "storage": "HDFS (Hadoop cluster)",
        "location": hv.get("hdfs_location", "hdfs://atc-hadoop-m01:8020/data/historical/sales_orders"),
        "namenode": HDFS_NN_URL,
        "metastore": "Hive Metastore (Derby)",
        "exec": hv.get("exec", "MapReduce"),
        "data_files": hv.get("hdfs_files"),
        "data_bytes": hv.get("hdfs_size_bytes"),
        "rows": hv.get("rows"),
        "measured_at": hv.get("measured_at"),
    }
    data = {"ok": True, "sources": [trino_src, hive_src], "hdfs_namenode": HDFS_NN_URL}
    return JSONResponse(_store("source", data))


# --------------------------------------------------------------------------- Hive (real, via SSH)
def _load_seed() -> dict[str, Any]:
    for p in (_PERSIST, _SEED):
        try:
            if p.exists():
                return json.loads(p.read_text())
        except Exception:
            pass
    return {}


_hive_cache: dict[str, Any] = {"data": _load_seed(), "ts": 0.0, "refreshing": False}


def _ssh_hive_bench() -> dict[str, Any]:
    import paramiko

    cli = paramiko.SSHClient()
    cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    cli.connect(HIVE_SSH_HOST, username=HIVE_SSH_USER, timeout=10, look_for_keys=True, allow_agent=False)
    try:
        _stdin, stdout, _stderr = cli.exec_command(HIVE_BENCH_CMD, timeout=120)
        out = stdout.read().decode("utf-8", "replace").strip()
    finally:
        cli.close()
    line = [ln for ln in out.splitlines() if ln.strip().startswith("{")]
    return json.loads(line[-1]) if line else {}


async def _refresh_hive() -> None:
    if _hive_cache["refreshing"]:
        return
    _hive_cache["refreshing"] = True
    try:
        res = await asyncio.to_thread(_ssh_hive_bench)
        if res and res.get("ok"):
            _hive_cache["data"] = res
            _hive_cache["ts"] = time.time()
            try:
                _PERSIST.write_text(json.dumps(res))
            except Exception:
                pass
    except Exception:
        pass
    finally:
        _hive_cache["refreshing"] = False


@router.get("/engines")
async def engines():
    # Trino measured live (same workload as Hive: revenue by region)
    bench_sql = f"SELECT region, sum(amount) AS revenue, count(*) AS n FROM {TABLE} GROUP BY region ORDER BY revenue DESC"
    trino_entry: dict[str, Any] = {"name": "Trino", "measured": True, "ok": False}
    try:
        await _trino(bench_sql)  # warm
        t0 = time.monotonic()
        res = await _trino(bench_sql)
        trino_ms = round((time.monotonic() - t0) * 1000)
        st = _trino_stats(res["stats"])
        trino_entry.update({
            "ok": True,
            "latency_ms": trino_ms,
            "model": "MPP · in-memory pipelined",
            "storage": "Iceberg / Parquet on S3",
            "best_for": "Interactive federated SQL & lakehouse BI",
            "note": "Live query on iceberg.hadoop.historical_sales",
            "processed_rows": st["processed_rows"],
            "processed_bytes": st["processed_bytes"],
            "by_region": [{"region": r[0], "revenue": float(r[1] or 0), "n": int(r[2] or 0)} for r in res["rows"]],
        })
    except Exception as e:
        trino_entry["error"] = str(e)
        trino_ms = 0

    # Hive: REAL measurement, cached + refreshed over SSH
    if time.time() - _hive_cache["ts"] > HIVE_BENCH_TTL:
        asyncio.create_task(_refresh_hive())
    hv = _hive_cache["data"] or {}
    hive_ms = int(hv.get("query_time_ms") or 0)
    hive_entry = {
        "name": "Hive",
        "measured": bool(hv.get("ok")),
        "ok": bool(hv.get("ok")),
        "latency_ms": hive_ms,
        "wall_ms": hv.get("wall_ms"),
        "model": f"Batch · {hv.get('exec', 'MapReduce')}",
        "storage": hv.get("storage", "HDFS · CSV (TEXTFILE)"),
        "best_for": "Large ETL / batch transforms",
        "note": f"Live measurement · Apache Hive {hv.get('version', '3.1.3')} on HDFS",
        "version": hv.get("version"),
        "measured_at": hv.get("measured_at"),
        "rows": hv.get("rows"),
        "by_region": hv.get("by_region", []),
    }

    # Impala: representative (not deployed on this cluster)
    base = trino_ms or 300
    impala_entry = {
        "name": "Impala",
        "measured": False,
        "ok": True,
        "latency_ms": max(1, round(base * 0.9)),
        "model": "MPP · C++ daemons (LLVM codegen)",
        "storage": "Parquet on HDFS / S3 (HMS)",
        "best_for": "Low-latency interactive BI on Hadoop",
        "note": "Representative — Impala daemons not deployed on this cluster",
    }

    data = {
        "ok": True,
        "benchmark": "Revenue by region over ~25k historical rows",
        "trino_sql": bench_sql,
        "hive_sql": hv.get("query"),
        "engines": [trino_entry, hive_entry, impala_entry],
    }
    return JSONResponse(data)