eSlider
ea062af86a
Explorer and exporter were scanning the full Parquet lake every 1–5s (~2000 files, 200MB+), driving ~2.2 CPU cores. Limit metrics to the last N flights, cache DuckDB views and the partition tree, slow live polls to 2s, and keep drones alive after seal to stop restart churn. Add node-exporter, scan-duration metrics, and a Swarm Platform Grafana dashboard for node CPU/memory and scan health.
44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
"""Scan the Hive-partitioned lake without re-reading every historical flight."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
|
|
def flight_window() -> int:
|
|
return max(1, int(os.environ.get("METRIC_FLIGHT_WINDOW", "5")))
|
|
|
|
|
|
def recent_flights(data_dir: Path, dataset: str, limit: int | None = None) -> list[Path]:
|
|
"""Most recently modified flight= partitions for a dataset."""
|
|
limit = limit or flight_window()
|
|
base = data_dir / f"dataset={dataset}"
|
|
if not base.is_dir():
|
|
return []
|
|
flights = [p for p in base.iterdir() if p.is_dir() and p.name.startswith("flight=")]
|
|
flights.sort(key=lambda p: p.stat().st_mtime, reverse=True)
|
|
return flights[:limit]
|
|
|
|
|
|
def parquet_reader(data_dir: Path, dataset: str, *, limit: int | None = None) -> str:
|
|
"""DuckDB read_parquet() source limited to recent flights."""
|
|
flights = recent_flights(data_dir, dataset, limit)
|
|
if not flights:
|
|
path = data_dir / f"dataset={dataset}" / "**" / "*.parquet"
|
|
return f"'{path}', hive_partitioning=true, union_by_name=true"
|
|
if len(flights) == 1:
|
|
return f"'{flights[0]}/**/*.parquet', hive_partitioning=true, union_by_name=true"
|
|
inner = ", ".join(f"'{f}/**/*.parquet'" for f in flights)
|
|
return f"[{inner}], hive_partitioning=true, union_by_name=true"
|
|
|
|
|
|
def iter_parquet_files(data_dir: Path, *, flight_limit: int | None = None) -> list[Path]:
|
|
"""Parquet paths under recent flights only — avoids full-lake rglob."""
|
|
limit = flight_limit or flight_window()
|
|
out: list[Path] = []
|
|
for ds in ("telemetry", "detections", "state"):
|
|
for flight in recent_flights(data_dir, ds, limit):
|
|
out.extend(flight.rglob("*.parquet"))
|
|
return out
|