From 71a64d5a2141544a7de1ae9b15fd0eaf8a607a58 Mon Sep 17 00:00:00 2001 From: mo Date: Fri, 26 Jun 2026 15:44:43 +0000 Subject: [PATCH] feat(hadoop): embedded lakehouse analytics + Impala/Hive vs Trino comparison Adds an Analytics sub-tab to the Hadoop view with live KPIs and revenue breakdowns (by year/region/category/channel) queried from Trino over iceberg.hadoop.historical_sales, plus a query-engine comparison panel. Trino latency is measured live; Impala and Hive are shown as clearly-labelled representative figures (those engines are not deployed). New cached endpoints /api/hadoop/analytics and /api/hadoop/engines. --- api/Dockerfile | 2 +- api/hadoop_analytics.py | 190 ++++++++++++++ api/main.py | 2 + .../components/features/HadoopAnalytics.tsx | 234 ++++++++++++++++++ ui/src/components/features/HdfsView.tsx | 18 +- 5 files changed, 443 insertions(+), 3 deletions(-) create mode 100644 api/hadoop_analytics.py create mode 100644 ui/src/components/features/HadoopAnalytics.tsx diff --git a/api/Dockerfile b/api/Dockerfile index db17ed5..f6a5037 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -4,7 +4,7 @@ WORKDIR /app RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py . +COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py . RUN mkdir -p /data ENV DATABASE_URL=sqlite:////data/atc-agents.db EXPOSE 3201 diff --git a/api/hadoop_analytics.py b/api/hadoop_analytics.py new file mode 100644 index 0000000..fb4e5ba --- /dev/null +++ b/api/hadoop_analytics.py @@ -0,0 +1,190 @@ +"""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) diff --git a/api/main.py b/api/main.py index 1f2d9a0..301a54a 100644 --- a/api/main.py +++ b/api/main.py @@ -39,6 +39,7 @@ from presentation_static import get_static_deck, list_static_decks from storage_s3 import router as storage_s3_router from hdfs_api import router as hdfs_router from pipeline_ops import router as pipeline_router +from hadoop_analytics import router as hadoop_router from elasticsearch_api import router as elasticsearch_router from sql_console import router as sql_router from ssh_terminal import ssh_session @@ -721,6 +722,7 @@ app = FastAPI(title="ATC Command Center API", lifespan=lifespan) app.include_router(storage_s3_router) app.include_router(hdfs_router) app.include_router(pipeline_router) +app.include_router(hadoop_router) app.include_router(elasticsearch_router) app.include_router(sql_router) app.add_middleware( diff --git a/ui/src/components/features/HadoopAnalytics.tsx b/ui/src/components/features/HadoopAnalytics.tsx new file mode 100644 index 0000000..904cb9b --- /dev/null +++ b/ui/src/components/features/HadoopAnalytics.tsx @@ -0,0 +1,234 @@ +import { useCallback, useEffect, useState } from 'react' +import { Activity, BarChart3, Database, Gauge, Layers, Loader2, RefreshCw, Zap } from 'lucide-react' +import { cn } from '../../lib/utils' +import { subTabIdle } from '../../lib/tabActive' + +type Kpis = { + orders: number + revenue: number + aov: number + units: number + regions: number + year_min: number | null + year_max: number | null +} + +type Analytics = { + ok: boolean + table?: string + location?: string + kpis?: Kpis + by_year?: { year: number; revenue: number; orders: number }[] + by_region?: { region: string; revenue: number }[] + by_category?: { category: string; revenue: number }[] + by_channel?: { channel: string; revenue: number; orders: number }[] + error?: string +} + +type Engine = { + name: string + measured: boolean + latency_ms: number + model: string + storage: string + best_for: string + note: string +} + +type Engines = { + ok: boolean + benchmark_sql?: string + rows_scanned?: number + measured_engine?: string + engines?: Engine[] + error?: string +} + +const usd = (n: number) => { + if (n >= 1e9) return `$${(n / 1e9).toFixed(2)}B` + if (n >= 1e6) return `$${(n / 1e6).toFixed(1)}M` + if (n >= 1e3) return `$${(n / 1e3).toFixed(0)}K` + return `$${n.toFixed(0)}` +} +const num = (n: number) => n.toLocaleString('en-US') + +const ENGINE_COLORS: Record = { + Trino: '#22d3ee', + Impala: '#fb923c', + Hive: '#a78bfa', +} + +function Kpi({ icon: Icon, label, value, sub }: { icon: typeof Gauge; label: string; value: string; sub?: string }) { + return ( +
+
+ {label} +
+
{value}
+ {sub &&
{sub}
} +
+ ) +} + +function BarRow({ label, value, max, display, color = '#34d399' }: { label: string; value: number; max: number; display: string; color?: string }) { + const pct = max > 0 ? Math.max(2, (value / max) * 100) : 0 + return ( +
+ {label} +
+
+
+ {display} +
+ ) +} + +function ChartCard({ title, icon: Icon, children }: { title: string; icon: typeof BarChart3; children: React.ReactNode }) { + return ( +
+

+ {title} +

+
{children}
+
+ ) +} + +export function HadoopAnalytics() { + const [data, setData] = useState(null) + const [engines, setEngines] = useState(null) + const [loading, setLoading] = useState(false) + + const load = useCallback(async () => { + setLoading(true) + try { + const [a, e] = await Promise.all([ + fetch('/api/hadoop/analytics').then((r) => r.json()), + fetch('/api/hadoop/engines').then((r) => r.json()), + ]) + setData(a) + setEngines(e) + } catch { + setData({ ok: false, error: 'Analytics API unavailable' }) + } finally { + setLoading(false) + } + }, []) + + useEffect(() => { + load() + }, [load]) + + const k = data?.kpis + const maxYear = Math.max(1, ...(data?.by_year || []).map((r) => r.revenue)) + const maxRegion = Math.max(1, ...(data?.by_region || []).map((r) => r.revenue)) + const maxCat = Math.max(1, ...(data?.by_category || []).map((r) => r.revenue)) + const maxChan = Math.max(1, ...(data?.by_channel || []).map((r) => r.revenue)) + const maxLat = Math.max(1, ...(engines?.engines || []).map((e) => e.latency_ms)) + + return ( +
+
+
+

Lakehouse Analytics

+

{data?.location || 'iceberg.hadoop.historical_sales'}

+
+ +
+ + {loading && !data && ( +

+ Querying Trino… +

+ )} + {data && !data.ok &&

{data.error}

} + + {k && ( + <> +
+ + + + + + +
+ +
+ + {(data.by_year || []).map((r) => ( + + ))} + + + {(data.by_region || []).map((r) => ( + + ))} + + + {(data.by_category || []).map((r) => ( + + ))} + + + {(data.by_channel || []).map((r) => ( + + ))} + +
+ + )} + + {/* Engine comparison */} +
+
+

+ Query engines · Impala & Hive vs Trino +

+ {engines?.rows_scanned ? ( + {num(engines.rows_scanned)} rows · lower = faster + ) : null} +
+

+ Same aggregation over the curated lakehouse table.{' '} + Trino is measured live; Impala & Hive are representative + reference figures (those engines are not deployed on this platform). +

+
+ {(engines?.engines || []).map((e) => ( +
+
+ {e.name} + + {e.measured ? 'live' : 'representative'} + +
+
+
+ {num(e.latency_ms)} ms +
+
+ Model: {e.model} + Storage: {e.storage} + Best for: {e.best_for} +
+
+ ))} +
+ {engines?.benchmark_sql && ( + 
{engines.benchmark_sql}
+ )} +
+
+ ) +} diff --git a/ui/src/components/features/HdfsView.tsx b/ui/src/components/features/HdfsView.tsx index 9ee1244..d1bd9f4 100644 --- a/ui/src/components/features/HdfsView.tsx +++ b/ui/src/components/features/HdfsView.tsx @@ -1,7 +1,8 @@ import { useCallback, useEffect, useState } from 'react' -import { ChevronRight, Download, Eye, FileText, Folder, Loader2, RefreshCw, Server, X } from 'lucide-react' +import { BarChart3, ChevronRight, Download, Eye, FileText, Folder, HardDrive, Loader2, RefreshCw, Server, X } from 'lucide-react' import { cn } from '../../lib/utils' import { subTabActive, subTabIdle } from '../../lib/tabActive' +import { HadoopAnalytics } from './HadoopAnalytics' type HdfsEntry = { name: string @@ -36,6 +37,7 @@ export function HdfsView() { const [error, setError] = useState(null) const [preview, setPreview] = useState<{ path: string; text: string; binary: boolean; truncated: boolean } | null>(null) const [previewing, setPreviewing] = useState(false) + const [tab, setTab] = useState<'files' | 'analytics'>('files') const loadHealth = useCallback(async () => { try { @@ -106,7 +108,15 @@ export function HdfsView() { {typeof health?.capacity_used_pct === 'number' ? ` (${health.capacity_used_pct}%)` : ''}

-
+
+
+ + +
NameNode UI @@ -116,6 +126,9 @@ export function HdfsView() {
+ {tab === 'analytics' && } + + {tab === 'files' && (
+ )} {preview && (
setPreview(null)}>