HJQ/tianxuan
2
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

perf: optimize for 25M rows on 8GB RAM

Browse Source 
- globe.py: add streaming=True to all lf.head().collect() calls (3 places)
- entities.py: chunked aggregation (100K batches) for datasets > 1M rows
- constants.py: reduce MAX_SAMPLE_ROWS 50000->30000 for 8GB RAM
- tools/__init__.py: add __all__ for '*' re-export (fix tool_registry import)
- clustering.py: fix misplaced logger/import traceback syntax errors
- analysis.py: fix indentation of pass statement
- anomalies.py: fix misplaced import traceback syntax error
This commit is contained in:
PM-pinou
2026-07-23 22:54:45 +08:00
parent 88b26f16fe
commit 9508cc8d58
7 changed files with 89 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
analysis/constants.py
+2 -2
View File
@@ -12,8 +12,8 @@ import polars as pl
# Clustering / UMAP
# ---------------------------------------------------------------------------
MAX_SAMPLE_ROWS = 50000
"""Maximum rows for clustering sample (downsampling threshold)."""
MAX_SAMPLE_ROWS = 30000
"""Maximum rows for clustering sample (downsampling threshold, tuned for 8GB RAM)."""
MAX_DISTRIBUTION_SAMPLE = 10000
"""Default sample size for distribution exploration and scoring."""
analysis/tools/__init__.py
+41
View File
@@ -60,3 +60,44 @@ from .data_mgmt import (
_handle_clone_dataset,
)
from .distance_matrix import _handle_compute_distance_matrix
__all__ = [
# Helpers
'_json_size', '_truncate_response', '_build_filter_expr', '_resolve_dataset',
'_count_rows', '_get_numeric_columns', 'NUMERIC_DTYPES', 'NUMERIC_TYPE_NAMES',
# Registry + dispatch
'get_tools_meta', 'handle_call',
# Load
'_handle_load_data',
# Profile
'_handle_profile_data',
# Filter
'_handle_filter_data',
# Preprocess
'_handle_preprocess_data',
# Clustering
'_handle_run_clustering', '_handle_filter_and_cluster',
# Evaluate
'_handle_evaluate_clustering',
# Features
'_handle_extract_features', '_save_features_to_db',
# Export
'_handle_export_results',
# Entities
'_handle_build_entity_profiles', '_handle_compute_scores', '_add_adaptive_scores',
'_ENTITY_KEYWORDS', '_NUMERIC_TYPES_FOR_AGG', '_ANOMALY_FEATURE_COLS',
# Anomalies
'_handle_detect_anomalies', '_handle_visualize_anomalies',
# Diagnostics
'_handle_validate_data', '_handle_explore_distributions', '_handle_find_outliers',
'_handle_diagnose_clustering', '_handle_compare_datasets',
'_handle_export_debug_sample', '_handle_repair_schema',
# Analysis
'_handle_analyze_patterns', '_handle_analyze_temporal', '_handle_analyze_fft',
'_handle_analyze_tls_health', '_handle_analyze_geo_distribution',
'_handle_analyze_entity_detail',
# Data management
'_handle_list_datasets', '_handle_drop_dataset', '_handle_clone_dataset',
# Distance matrix
'_handle_compute_distance_matrix',
]
analysis/tools/analysis.py
+1 -1
View File
@@ -285,7 +285,7 @@ async def _handle_analyze_geo_distribution(dataset_id: str) -> dict:
result['international_pct'] = round(100.0 - result['domestic_pct'], 1)
except Exception:
logger.error("_handle_analyze_geo_distribution failed: {}".format(traceback.format_exc()))
pass
pass
return result
analysis/tools/anomalies.py
+1 -1
View File
@@ -1,4 +1,5 @@
"""Anomaly detection + visualization handlers."""
import traceback
import numpy as np
from analysis.constants import MAX_DISTRIBUTION_SAMPLE, RANDOM_SEED
import polars as pl
@@ -12,7 +13,6 @@ logger = logging.getLogger(__name__)
async def _handle_detect_anomalies(dataset_id: str, contamination: float = 0.05) -> dict:
"""Run Isolation Forest on scored entity data."""
from sklearn.ensemble import IsolationForest
import traceback
entry, err = _resolve_dataset(dataset_id)
if err:
return err
analysis/tools/clustering.py
+2 -2
View File
@@ -6,6 +6,7 @@ Contains:
- _cluster_core: shared clustering pipeline extracted from both handlers
"""
import logging
import traceback
from analysis.constants import LOW_MEMORY_THRESHOLD_GB, LOW_MEMORY_THRESHOLD_ROWS, MAX_CLUSTER_FEATURES, MAX_SAMPLE_ROWS, RANDOM_SEED
from typing import Optional
@@ -14,13 +15,13 @@ import polars as pl
from ..session_store import SessionStore, _generate_id
from ._helpers import (
logger = logging.getLogger(__name__)
NUMERIC_DTYPES,
_build_filter_expr,
_resolve_dataset,
_count_rows,
)
logger = logging.getLogger(__name__)
_clust_logger = logging.getLogger(__name__)
@@ -172,7 +173,6 @@ def _cluster_core(
quality['davies_bouldin_score'] = None
try:
from sklearn.metrics import calinski_harabasz_score
import traceback
chs = calinski_harabasz_score(data_scaled, labels_array)
quality['calinski_harabasz_score'] = round(float(chs), 4)
except Exception:
analysis/tools/entities.py
+39 -1
View File
@@ -116,7 +116,45 @@ async def _handle_build_entity_profiles(
agg_exprs.append(pl.col(c).n_unique().alias(alias_name))
# ── Execute grouping ───────────────────────────────────────────────
entity_lf = lf.group_by(group_cols).agg(agg_exprs)
row_count = _count_rows(lf)
ENTITY_CHUNK_THRESHOLD = 1_000_000 # 1M rows
ENTITY_CHUNK_SIZE = 100_000 # 100K rows per chunk
if row_count >= ENTITY_CHUNK_THRESHOLD:
# Chunked path: process in 100K batches → group → merge across chunks
all_chunk_dfs = []
for offset in range(0, row_count, ENTITY_CHUNK_SIZE):
chunk_df = lf.slice(offset, ENTITY_CHUNK_SIZE).collect(streaming=True)
chunk_grouped = chunk_df.group_by(group_cols).agg(agg_exprs)
all_chunk_dfs.append(chunk_grouped)
merged_df = pl.concat(all_chunk_dfs)
# Re-aggregate across chunks (same entity may appear in multiple chunks)
re_agg_exprs: list[pl.Expr] = []
for col_name in merged_df.columns:
if col_name in group_cols:
continue
if col_name == 'flow_count':
re_agg_exprs.append(pl.sum('flow_count').alias('flow_count'))
elif col_name.startswith('total_'):
re_agg_exprs.append(pl.sum(col_name).alias(col_name))
elif col_name.startswith('avg_'):
# Weighted average: sum(total) / sum(flow_count)
base = col_name[4:]
total_col = f'total_{base}'
if total_col in merged_df.columns:
re_agg_exprs.append(
(pl.sum(total_col) / pl.sum('flow_count')).alias(col_name))
elif col_name.startswith('unique_'):
# Approximate: max of per-chunk distinct counts
re_agg_exprs.append(pl.max(col_name).alias(col_name))
entity_lf = merged_df.lazy().group_by(group_cols).agg(re_agg_exprs)
logger.info('[ENTITY] chunked aggregation: %d rows → %d chunks → %d merged rows',
row_count, len(all_chunk_dfs), len(merged_df))
else:
entity_lf = lf.group_by(group_cols).agg(agg_exprs)
n_entities = _count_rows(entity_lf)
# ── Store result ───────────────────────────────────────────────────
analysis/views/globe.py
+3 -3
View File
@@ -130,7 +130,7 @@ def globe_view(request):
entry = store.get_dataset(data_param)
if entry is not None:
lf = entry['lazyframe']
df = lf.head(MAX_ROWS_PER_RUN).collect()
df = lf.head(MAX_ROWS_PER_RUN).collect(streaming=True)
flows = _extract_flows_from_df(df, MAX_ROWS_PER_RUN)
except Exception as exc:
logging.getLogger(__name__).warning(f'globe ?data= param failed: {exc}')
@@ -172,7 +172,7 @@ def globe_view(request):
lf = load_from_db(run.sqlite_table)
if lf is None:
lf, schema, rc, fc, mem = load_csv_directory(run.csv_glob)
df = lf.head(MAX_ROWS_PER_RUN).collect()
df = lf.head(MAX_ROWS_PER_RUN).collect(streaming=True)
flows.extend(_extract_flows_from_df(df, MAX_ROWS_PER_RUN))
all_selected_failed = False
break # stop after first successful load
@@ -193,7 +193,7 @@ def globe_view(request):
if not latest.csv_glob:
continue
lf, schema, rc, fc, mem = load_csv_directory(latest.csv_glob)
df = lf.head(MAX_ROWS_PER_RUN).collect()
df = lf.head(MAX_ROWS_PER_RUN).collect(streaming=True)
flows.extend(_extract_flows_from_df(df, MAX_ROWS_PER_RUN))
selected_runs = [latest]
break