mo
96d490807a
Deploys Apache Hive 3.1.3 (Derby metastore + external table over the HDFS historical CSV, MapReduce exec) on the Hadoop master, so the engine comparison shows a REAL measured Hive latency (~5.2s) next to live Trino (~0.3s); Impala stays clearly-labelled representative. The API re-measures Hive over SSH on a 30-min TTL (cached + persisted, with a committed seed). Adds filters (year/region/category/channel), a region×category heatmap, Trino exec stats, and a "where is this data read from" lineage panel (Trino->S3/Iceberg/Parquet with snapshot+files, Hive->HDFS/CSV with namenode+files). Mounts host SSH key read-only into the api container for the live Hive benchmark.
363 lines
14 KiB
Python
363 lines
14 KiB
Python
"""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)
|