"""Hadoop lakehouse analytics for the Command Center 'Hadoop' tab. Runs live analytical queries on the curated lakehouse table (iceberg.hadoop.historical_sales, stored as Parquet on Dell ECS S3) via Trino, and exposes a query-engine comparison: Trino is measured live, while Impala and Hive are shown as clearly-labelled *representative* figures (those engines are not deployed on this platform). Results are cached briefly so the UI stays snappy. """ from __future__ import annotations import asyncio import time from typing import Any import httpx from fastapi import APIRouter from fastapi.responses import JSONResponse import os 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" router = APIRouter(prefix="/api/hadoop", tags=["hadoop"]) _cache: dict[str, Any] = {} _TTL = 60.0 # seconds async def _trino(sql: str, deadline_s: float = 30.0) -> list[list[Any]]: """Execute a Trino statement and return all rows (list of lists).""" headers = { "X-Trino-User": TRINO_USER, "X-Trino-Catalog": "iceberg", "X-Trino-Schema": "hadoop", } rows: list[list[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: err = payload.get("error") if err: raise RuntimeError(err.get("message", str(err))) rows.extend(payload.get("data", []) or []) nxt = payload.get("nextUri") if not nxt: break if time.monotonic() - start > deadline_s: raise TimeoutError("Trino query exceeded deadline") await asyncio.sleep(0.05) rr = await client.get(nxt, headers={"X-Trino-User": TRINO_USER}) rr.raise_for_status() payload = rr.json() return rows def _cached(key: str): 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("/analytics") async def analytics(): cached = _cached("analytics") if cached is not None: return JSONResponse(cached) try: kpis_q = ( f"SELECT count(*) AS orders, sum(amount) AS revenue, avg(amount) AS aov, " f"sum(quantity) AS units, count(DISTINCT region) AS regions, " f"min(order_year) AS min_y, max(order_year) AS max_y FROM {TABLE}" ) by_year_q = ( f"SELECT order_year, sum(amount) AS revenue, count(*) AS orders " f"FROM {TABLE} GROUP BY order_year ORDER BY order_year" ) by_region_q = ( f"SELECT region, sum(amount) AS revenue FROM {TABLE} " f"GROUP BY region ORDER BY revenue DESC" ) by_category_q = ( f"SELECT product_category, sum(amount) AS revenue FROM {TABLE} " f"GROUP BY product_category ORDER BY revenue DESC" ) by_channel_q = ( f"SELECT channel, sum(amount) AS revenue, count(*) AS orders FROM {TABLE} " f"GROUP BY channel ORDER BY revenue DESC" ) kpis, by_year, by_region, by_category, by_channel = await asyncio.gather( _trino(kpis_q), _trino(by_year_q), _trino(by_region_q), _trino(by_category_q), _trino(by_channel_q), ) k = kpis[0] if kpis else [0, 0, 0, 0, 0, None, None] data = { "ok": True, "table": TABLE, "location": "s3://data/hadoop/historical_sales (Iceberg/Parquet on Dell ECS)", "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], }, "by_year": [{"year": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_year], "by_region": [{"region": r[0], "revenue": float(r[1] or 0)} for r in by_region], "by_category": [{"category": r[0], "revenue": float(r[1] or 0)} for r in by_category], "by_channel": [{"channel": r[0], "revenue": float(r[1] or 0), "orders": int(r[2] or 0)} for r in by_channel], } return JSONResponse(_store("analytics", data)) except Exception as e: return JSONResponse({"ok": False, "error": str(e)}, status_code=200) @router.get("/engines") async def engines(): """Live-measured Trino latency vs representative Impala/Hive figures.""" cached = _cached("engines") if cached is not None: return JSONResponse(cached) bench_sql = ( f"SELECT region, product_category, sum(amount) AS revenue, " f"avg(unit_price) AS avg_price, count(*) AS n " f"FROM {TABLE} GROUP BY region, product_category ORDER BY revenue DESC" ) try: # warm + measure (best of 2 to dampen JIT/scheduling noise) await _trino(bench_sql) t0 = time.monotonic() await _trino(bench_sql) trino_ms = round((time.monotonic() - t0) * 1000) cnt = await _trino(f"SELECT count(*) FROM {TABLE}") rows_scanned = int(cnt[0][0]) if cnt else 0 # Representative multipliers for an analytical aggregate over columnar # Parquet of this size. Impala (MPP C++ daemons) is in the same league # as Trino; Hive (Tez/MR batch) pays heavy job-startup cost. data = { "ok": True, "benchmark_sql": bench_sql, "rows_scanned": rows_scanned, "measured_engine": "Trino", "engines": [ { "name": "Trino", "measured": 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", }, { "name": "Impala", "measured": False, "latency_ms": max(1, round(trino_ms * 0.9)), "model": "MPP · C++ daemons (LLVM codegen)", "storage": "Parquet on HDFS / S3 (HMS)", "best_for": "Low-latency interactive BI on Hadoop", "note": "Representative — no Impala daemon deployed", }, { "name": "Hive", "measured": False, "latency_ms": max(1, round(trino_ms * 8)), "model": "Batch · Tez / MapReduce", "storage": "ORC / Parquet on HDFS (HMS)", "best_for": "Large ETL / batch transforms", "note": "Representative — no HiveServer2 deployed", }, ], } return JSONResponse(_store("engines", data)) except Exception as e: return JSONResponse({"ok": False, "error": str(e)}, status_code=200)