diff --git a/api/Dockerfile b/api/Dockerfile index f6a5037..cf89312 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 hadoop_analytics.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 hive_bench_seed.json . 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 index fb4e5ba..631a01b 100644 --- a/api/hadoop_analytics.py +++ b/api/hadoop_analytics.py @@ -1,67 +1,113 @@ """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. +- /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 +from fastapi import APIRouter, Query 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" +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 = 60.0 # seconds +_TTL = 45.0 + +# dimension -> Iceberg column +DIMS = {"years": "order_year", "regions": "region", "categories": "product_category", "channels": "channel"} -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", - } +# --------------------------------------------------------------------------- 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: - err = payload.get("error") - if err: - raise RuntimeError(err.get("message", str(err))) + 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.05) + 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 + return {"rows": rows, "columns": cols or [], "stats": stats} -def _cached(key: str): +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): + if item and (time.time() - item["ts"] < ttl): return item["data"] return None @@ -71,42 +117,62 @@ def _store(key: str, data: Any): 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(): - cached = _cached("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(*) 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}" + 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}" ) - 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" + 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"), ) - 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)", + "filtered": bool(where), + "where": where, "kpis": { "orders": int(k[0] or 0), "revenue": float(k[1] or 0), @@ -115,76 +181,182 @@ async def analytics(): "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], - "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], + "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("analytics", data)) + return JSONResponse(_store(key, data)) except Exception as e: - return JSONResponse({"ok": False, "error": str(e)}, status_code=200) + 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(): - """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" - ) + # 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: - # warm + measure (best of 2 to dampen JIT/scheduling noise) - await _trino(bench_sql) + await _trino(bench_sql) # warm t0 = time.monotonic() - await _trino(bench_sql) + res = 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 = { + st = _trino_stats(res["stats"]) + trino_entry.update({ "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)) + "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: - return JSONResponse({"ok": False, "error": str(e)}, status_code=200) + 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) diff --git a/api/hive_bench_seed.json b/api/hive_bench_seed.json new file mode 100644 index 0000000..0438925 --- /dev/null +++ b/api/hive_bench_seed.json @@ -0,0 +1 @@ +{"engine":"Hive","ok":true,"attempts":1,"version":"3.1.3","exec":"MapReduce (local)","storage":"HDFS · CSV (TEXTFILE)","table":"lake.historical_sales","hdfs_location":"hdfs://atc-hadoop-m01:8020/data/historical/sales_orders","hdfs_size_bytes":2496372,"hdfs_files":5,"query":"SELECT region, round(sum(amount),2) AS revenue, count(*) AS n FROM lake.historical_sales GROUP BY region ORDER BY revenue DESC","query_time_ms":5340,"wall_ms":14440,"rows":25000,"measured_at":"2026-06-26T17:00:37Z","by_region":[{"region":"LATAM","revenue":2.549512032E7,"n":5050},{"region":"EMEA","revenue":2.525153502E7,"n":5047},{"region":"NA","revenue":2.516295049E7,"n":5021},{"region":"EU","revenue":2.489941114E7,"n":4976},{"region":"APAC","revenue":2.474656452E7,"n":4906}]} diff --git a/docker-compose.yml b/docker-compose.yml index 840b13d..c68d62c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -53,6 +53,7 @@ services: ELASTIC_PASSWORD: ${ELASTIC_PASSWORD:-} volumes: - api_data:/data + - /root/.ssh:/root/.ssh:ro depends_on: redis: condition: service_started diff --git a/infra/hadoop/hive_bench.sh b/infra/hadoop/hive_bench.sh new file mode 100644 index 0000000..b0af712 --- /dev/null +++ b/infra/hadoop/hive_bench.sh @@ -0,0 +1,36 @@ +#!/bin/bash +# Real Hive benchmark on the HDFS historical dataset. Emits one JSON line. +# Retries because Hive's LocalJobRunner can flake on a cold first job. +export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk +export HIVE_HOME=/opt/hive +export HADOOP_HOME=/usr/lib/hadoop +export HADOOP_CONF_DIR=/opt/hive/hadoopconf +export HADOOP_USER_NAME=hdfs +export HADOOP_CLIENT_OPTS="-Xmx3g" +export PATH=$JAVA_HOME/bin:$HIVE_HOME/bin:$PATH + +Q="SELECT region, round(sum(amount),2) AS revenue, count(*) AS n FROM lake.historical_sales GROUP BY region ORDER BY revenue DESC" +ISO=$(date -u +%Y-%m-%dT%H:%M:%SZ) +HSIZE=$(hdfs dfs -du -s /data/historical/sales_orders 2>/dev/null | awk '{print $1}'); [ -z "$HSIZE" ] && HSIZE=0 +HFILES=$(hdfs dfs -ls -R /data/historical/sales_orders 2>/dev/null | grep -c "part-"); [ -z "$HFILES" ] && HFILES=0 + +ERR=$(mktemp); OUT=$(mktemp); RC=1; WALL=0; TT=0 +for attempt in 1 2 3; do + T0=$(date +%s%3N) + $HIVE_HOME/bin/hive -e "$Q" >"$OUT" 2>"$ERR" + RC=$? + T1=$(date +%s%3N) + WALL=$((T1-T0)) + if [ $RC -eq 0 ] && [ -s "$OUT" ]; then + TT=$(grep -oE "Time taken: [0-9.]+ seconds" "$ERR" | grep -oE "[0-9.]+" | sort -nr | head -1) + break + fi + sleep 2 +done +[ -z "$TT" ] && TT=0 +QMS=$(awk "BEGIN{printf \"%d\", $TT*1000}") +ROWS_JSON=$(awk -F'\t' 'NF>=3{printf "%s{\"region\":\"%s\",\"revenue\":%s,\"n\":%s}", (c++? ",":""), $1,$2,$3}' "$OUT") +cat </dev/null || cp -f /usr/lib/hadoop/client/guava-27.0-jre.jar /opt/hive/lib/ +ls /opt/hive/lib/guava-*.jar + +echo "=== hive-site.xml ===" +cat > /opt/hive/conf/hive-site.xml <<'XML' + + + + javax.jdo.option.ConnectionURLjdbc:derby:;databaseName=/opt/hive/metastore_db;create=true + javax.jdo.option.ConnectionDriverNameorg.apache.derby.jdbc.EmbeddedDriver + hive.metastore.warehouse.dir/user/hive/warehouse + hive.execution.enginemr + mapreduce.framework.namelocal + hive.exec.mode.local.autotrue + hive.exec.submitviachildfalse + hive.metastore.schema.verificationfalse + datanucleus.schema.autoCreateAlltrue + hive.server2.enable.doAsfalse + hive.stats.autogatherfalse + hive.metastore.event.db.notification.api.authfalse + +XML + +cat > /opt/hive/conf/hive-env.sh < /opt/hive/runhive.sh </tmp/schematool.log 2>&1 && echo "schema init OK" || { echo "schema init FAILED"; tail -25 /tmp/schematool.log; exit 1; } diff --git a/ui/src/components/features/HadoopAnalytics.tsx b/ui/src/components/features/HadoopAnalytics.tsx index 904cb9b..86901be 100644 --- a/ui/src/components/features/HadoopAnalytics.tsx +++ b/ui/src/components/features/HadoopAnalytics.tsx @@ -1,62 +1,52 @@ -import { useCallback, useEffect, useState } from 'react' -import { Activity, BarChart3, Database, Gauge, Layers, Loader2, RefreshCw, Zap } from 'lucide-react' +import { useCallback, useEffect, useMemo, useState } from 'react' +import { + Activity, BarChart3, Boxes, Cloud, Database, Gauge, GitBranch, HardDrive, + Layers, Loader2, RefreshCw, Server, Tag, Zap, +} from 'lucide-react' import { cn } from '../../lib/utils' -import { subTabIdle } from '../../lib/tabActive' +import { subTabActive, subTabIdle } from '../../lib/tabActive' type Kpis = { - orders: number - revenue: number - aov: number - units: number - regions: number - year_min: number | null - year_max: number | null + orders: number; revenue: number; aov: number; units: number; regions: number + year_min: number | null; year_max: number | null; avg_unit_price: number } - type Analytics = { - ok: boolean - table?: string - location?: string - kpis?: Kpis + ok: boolean; filtered?: boolean; where?: string; kpis?: Kpis by_year?: { year: number; revenue: number; orders: number }[] - by_region?: { region: string; revenue: number }[] - by_category?: { category: string; revenue: number }[] + by_region?: { region: string; revenue: number; orders: number }[] + by_category?: { category: string; revenue: number; orders: number }[] by_channel?: { channel: string; revenue: number; orders: number }[] + matrix?: { region: string; category: string; revenue: number }[] + trino_stats?: { processed_rows?: number; processed_bytes?: number; elapsed_ms?: number; cpu_ms?: number; peak_memory_bytes?: number; splits?: number } error?: string } - +type Filters = { ok: boolean; years: number[]; regions: string[]; categories: string[]; channels: string[] } +type SourceItem = { + engine: string; catalog?: string; schema?: string; database?: string; table: string + format: string; storage: string; location: string; s3_endpoint?: string; namenode?: string + metastore?: string; exec?: string; data_files?: number; data_bytes?: number; rows?: number + snapshot_id?: string; snapshot_at?: string; measured_at?: string; error?: string +} +type Source = { ok: boolean; sources?: SourceItem[]; hdfs_namenode?: string } type Engine = { - name: string - measured: boolean - latency_ms: number - model: string - storage: string - best_for: string - note: string + name: string; measured: boolean; ok: boolean; latency_ms: number; wall_ms?: number + model: string; storage: string; best_for: string; note: string + processed_rows?: number; processed_bytes?: number; version?: string; measured_at?: string + by_region?: { region: string; revenue: number; n: number }[] } +type Engines = { ok: boolean; benchmark?: string; trino_sql?: string; hive_sql?: string; engines?: Engine[] } -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', +const usd = (n: number) => (n >= 1e9 ? `$${(n / 1e9).toFixed(2)}B` : n >= 1e6 ? `$${(n / 1e6).toFixed(1)}M` : n >= 1e3 ? `$${(n / 1e3).toFixed(0)}K` : `$${n.toFixed(0)}`) +const num = (n?: number | null) => (n == null ? '—' : n.toLocaleString('en-US')) +const bytes = (n?: number | null) => { + if (n == null) return '—' + if (n >= 1e9) return `${(n / 1e9).toFixed(2)} GB` + if (n >= 1e6) return `${(n / 1e6).toFixed(2)} MB` + if (n >= 1e3) return `${(n / 1e3).toFixed(1)} KB` + return `${n} B` } +const ms = (n?: number | null) => (n == null ? '—' : n >= 1000 ? `${(n / 1000).toFixed(2)} s` : `${Math.round(n)} ms`) +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 ( @@ -70,7 +60,7 @@ function Kpi({ icon: Icon, label, value, sub }: { icon: typeof Gauge; label: str ) } -function BarRow({ label, value, max, display, color = '#34d399' }: { label: string; value: number; max: number; display: string; color?: string }) { +function BarRow({ label, value, max, display, sub, color = '#34d399' }: { label: string; value: number; max: number; display: string; sub?: string; color?: string }) { const pct = max > 0 ? Math.max(2, (value / max) * 100) : 0 return (
@@ -78,7 +68,7 @@ function BarRow({ label, value, max, display, color = '#34d399' }: { label: stri
- {display} + {display}{sub && · {sub}}
) } @@ -94,20 +84,39 @@ function ChartCard({ title, icon: Icon, children }: { title: string; icon: typeo ) } +function Chip({ active, onClick, children }: { active: boolean; onClick: () => void; children: React.ReactNode }) { + return ( + + ) +} + export function HadoopAnalytics() { + const [filters, setFilters] = useState(null) const [data, setData] = useState(null) + const [source, setSource] = useState(null) const [engines, setEngines] = useState(null) const [loading, setLoading] = useState(false) + const [selYears, setSelYears] = useState([]) + const [selRegions, setSelRegions] = useState([]) + const [selCats, setSelCats] = useState([]) + const [selChannels, setSelChannels] = useState([]) - const load = useCallback(async () => { + const qs = useMemo(() => { + const p = new URLSearchParams() + if (selYears.length) p.set('years', selYears.join(',')) + if (selRegions.length) p.set('regions', selRegions.join(',')) + if (selCats.length) p.set('categories', selCats.join(',')) + if (selChannels.length) p.set('channels', selChannels.join(',')) + return p.toString() + }, [selYears, selRegions, selCats, selChannels]) + + const loadAnalytics = useCallback(async (query: string) => { 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()), - ]) + const a = await fetch(`/api/hadoop/analytics${query ? `?${query}` : ''}`).then((r) => r.json()) setData(a) - setEngines(e) } catch { setData({ ok: false, error: 'Analytics API unavailable' }) } finally { @@ -115,9 +124,24 @@ export function HadoopAnalytics() { } }, []) - useEffect(() => { - load() - }, [load]) + const loadSideband = useCallback(async () => { + try { + const [f, s, e] = await Promise.all([ + fetch('/api/hadoop/filters').then((r) => r.json()), + fetch('/api/hadoop/source').then((r) => r.json()), + fetch('/api/hadoop/engines').then((r) => r.json()), + ]) + setFilters(f); setSource(s); setEngines(e) + } catch { /* ignore */ } + }, []) + + useEffect(() => { loadSideband() }, [loadSideband]) + useEffect(() => { loadAnalytics(qs) }, [qs, loadAnalytics]) + + const toggle = (v: T, list: T[], set: (x: T[]) => void) => + set(list.includes(v) ? list.filter((x) => x !== v) : [...list, v]) + const reset = () => { setSelYears([]); setSelRegions([]); setSelCats([]); setSelChannels([]) } + const anyFilter = selYears.length + selRegions.length + selCats.length + selChannels.length > 0 const k = data?.kpis const maxYear = Math.max(1, ...(data?.by_year || []).map((r) => r.revenue)) @@ -125,110 +149,166 @@ export function HadoopAnalytics() { 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)) + const matrixRegions = useMemo(() => [...new Set((data?.matrix || []).map((m) => m.region))], [data]) + const matrixCats = useMemo(() => [...new Set((data?.matrix || []).map((m) => m.category))], [data]) + const matrixMax = Math.max(1, ...(data?.matrix || []).map((m) => m.revenue)) + const matrixVal = (r: string, c: string) => data?.matrix?.find((m) => m.region === r && m.category === c)?.revenue || 0 return (
-
-
-

Lakehouse Analytics

-

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

+ {/* Filter bar */} +
+
+ Filters +
+ {anyFilter && } + +
+
+
+ toggle(Number(v), selYears, setSelYears)} /> + toggle(v, selRegions, setSelRegions)} /> + toggle(v, selCats, setSelCats)} /> + toggle(v, selChannels, setSelChannels)} />
-
- {loading && !data && ( -

- Querying Trino… -

- )} + {loading && !data &&

Querying Trino…

} {data && !data.ok &&

{data.error}

} {k && ( <> -
+
+ - +
- {(data.by_year || []).map((r) => ( - - ))} + {(data.by_year || []).map((r) => )} - {(data.by_region || []).map((r) => ( - - ))} + {(data.by_region || []).map((r) => )} - {(data.by_category || []).map((r) => ( - - ))} + {(data.by_category || []).map((r) => )} - {(data.by_channel || []).map((r) => ( - - ))} + {(data.by_channel || []).map((r) => )}
+ + {/* Region x Category heatmap */} + {matrixRegions.length > 0 && ( +
+

Revenue · region × category

+
+ + {matrixCats.map((c) => )} + + {matrixRegions.map((r) => ( + + + {matrixCats.map((c) => { + const v = matrixVal(r, c) + const a = Math.max(0.06, v / matrixMax) + return + })} + + ))} + +
Region \ Category{c}
{r}{usd(v)}
+
+
+ )} )} + {/* Data lineage: where is this read from */} +
+

Where is this data read from?

+
+ {(source?.sources || []).map((s) => { + const isTrino = s.engine === 'Trino' + return ( +
+
+ {isTrino ? : } {s.engine} → {s.storage} +
+
+
Table
{(s.catalog || s.database)}.{s.schema ? `${s.schema}.` : ''}{s.table}
+
Format
{s.format}
+
Location
{s.location}
+ {isTrino ? <>
S3 endpoint
{s.s3_endpoint}
: <>
NameNode
{s.namenode}
} +
Metastore
{s.metastore}
+
Data files
{num(s.data_files)} · {bytes(s.data_bytes)}
+
Rows
{num(s.rows)}
+ {s.snapshot_id && <>
Snapshot
{s.snapshot_id} @ {s.snapshot_at}
} + {s.exec && <>
Exec
{s.exec}
} +
+ {s.error &&

{s.error}

} +
+ ) + })} +
+
+ {/* Engine comparison */}
-

- Query engines · Impala & Hive vs Trino -

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

Query engines · Impala & Hive vs Trino

+ {engines?.benchmark} · lower = faster

- 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). + Trino (S3/Iceberg) and Hive (HDFS/CSV, Apache Hive 3.1.3 · MapReduce) are both measured live; + Impala is a representative reference (daemons not deployed on this cluster).

{(engines?.engines || []).map((e) => (
{e.name} - - {e.measured ? 'live' : 'representative'} - + {e.measured ? 'live' : 'representative'}
-
+
- {num(e.latency_ms)} ms + {ms(e.latency_ms)}
-
+
Model: {e.model} Storage: {e.storage} - Best for: {e.best_for} + {e.processed_rows != null && Scanned: {num(e.processed_rows)} rows · {bytes(e.processed_bytes)}} + {e.version && Engine: v{e.version}{e.wall_ms ? ` · wall ${ms(e.wall_ms)}` : ''}} + {e.measured_at && Measured: {e.measured_at.replace('T', ' ').replace('Z', ' UTC')}} + {e.note}
))}
- {engines?.benchmark_sql && ( - 
{engines.benchmark_sql}
- )} + {engines?.trino_sql && 
Trino:  {engines.trino_sql}{engines.hive_sql ? `\nHive:   ${engines.hive_sql}` : ''}
}
) } + +function FilterRow({ label, items, sel, onToggle }: { label: string; items: string[]; sel: string[]; onToggle: (v: string) => void }) { + if (!items.length) return null + return ( +
+ {label} +
+ {items.map((it) => onToggle(it)}>{it})} +
+
+ ) +} + +function Dt({ children }: { children: React.ReactNode }) { return
{children}
} +function Dd({ children, mono }: { children: React.ReactNode; mono?: boolean }) { return
{children}
}