From 3022b915ab4def09690bf4262708ebbdd6bed663 Mon Sep 17 00:00:00 2001 From: Michael Granitzer Date: Sun, 4 Jan 2026 14:36:21 +0100 Subject: [PATCH] feat: Add query stats command for dataset statistics - New query stats command calculates statistics via DuckDB queries - Shows record count, URL suffix/domain distribution, date range, MIME types - Visible progress bar with file count and records processed - Supports global --format flag (table, json, jsonl) - Fixed progress bar double-counting in stats command - Updated integration tests to use fixed dataset ID - Updated documentation with stats and aggregate examples - Replace deprecated lrz:latest with lexis:latest in docs --- docs/source/commands.md | 42 +++- owilix/cli/query.py | 45 +++- owilix/core/tasks/query.py | 202 ++++++++++++++++++ .../owilix/cli/test_query_less_integration.py | 8 +- 4 files changed, 281 insertions(+), 16 deletions(-) diff --git a/docs/source/commands.md b/docs/source/commands.md index 1457696..93c8584 100644 --- a/docs/source/commands.md +++ b/docs/source/commands.md @@ -181,27 +181,47 @@ owi --format jsonl -o results.jsonl query less -R lrz:latest --select "url,title ```bash # Filter by domains in file -owi query sites urls.txt -R lrz:latest +owi query sites urls.txt -R lexis:latest # JSON output with specific columns -owi --format jsonl query sites sites.csv -R all:latest/collectionName=cefal --select "url,title" +owi --format jsonl query sites sites.csv -R all/id=0350fecc-e58b-11f0-a8c9-8ebf6bb2cab9 --select "url,title" # Output to file with progress -owi --format jsonl+ -o output.jsonl query sites sites.csv -R lrz:latest +owi --format jsonl+ -o output.jsonl query sites sites.csv -R lexis:latest ``` -# Sites (URL Filtering) -owi query sites urls.txt -L main:latest -owi query sites domains.txt -R lrz:latest --json +##### query stats - Dataset Statistics -# Aggregation and Analysis -owi query aggregate -R lrz:latest --groupby url_suffix -owi query analyze my_job --status error +Calculates statistics over datasets using DuckDB queries: record counts, URL distributions, date ranges, MIME types. -# Creating new datasets -owi query slice -L main:latest -R "" --where "url LIKE '%example%'" --collection my_slice +```bash +# Table output with progress +owi query stats -R lexis:latest/collectionName=cefal + +# JSON output +owi --format json query stats -R all/id=0350fecc-e58b-11f0-a8c9-8ebf6bb2cab9 -o stats.json + +# Adjust top-k items +owi query stats -R lexis:latest --topk 50 ``` +##### query aggregate - SQL Aggregation + +Run SQL aggregation queries with GROUP BY support. + +```bash +# Count URLs by suffix +owi query aggregate -R lexis:latest --select "url_suffix,count(*) as c" --groupby url_suffix + +# With custom final query +owi query aggregate -R lexis:latest --select "url_domain,count(*) as cnt" --groupby url_domain \ + --aggregate "SELECT * FROM aggregates ORDER BY cnt DESC LIMIT 10" + +# JSON output +owi --format jsonl query aggregate -R lexis:latest --select "mime_type,count(*) as c" --groupby mime_type +``` + + #### Remote Commands Remote commands operate on the datacenters specified. diff --git a/owilix/cli/query.py b/owilix/cli/query.py index 74977bc..77e66f1 100644 --- a/owilix/cli/query.py +++ b/owilix/cli/query.py @@ -17,7 +17,7 @@ from owilix.cli._common.context import CLIContext from owilix.cli._common.executor import execute_command from owilix.core.tasks.query import ( query_less, query_sites, query_analyze, query_aggregate, - slice_query, stream_query + slice_query, stream_query, query_stats ) from owilix.core.tasks.warc.query_warc import warc as query_warc @@ -223,6 +223,49 @@ def aggregate( raise typer.Exit(code=1) +@app.command() +def stats( + ctx: typer.Context, + local: str = typer.Option("", "--local", "-L", help="Local dataset specifier"), + remote: str = typer.Option("", "--remote", "-R", help="Remote dataset specifier"), + files: str = typer.Option("**/*.parquet", "--files", "-f", help="File glob pattern"), + topk: int = typer.Option(20, "--topk", "-k", help="Number of top items per category"), + write_back: bool = typer.Option(False, "--write-back", "-w", help="Save stats to dataset .stats folder"), + force_write: bool = typer.Option(False, "--force", help="Force overwrite existing stats"), +): + """ + Calculate statistics over datasets. + + Statistics include record counts, URL distributions, date ranges, and MIME types. + Uses global --format flag for output (table, json, jsonl). + + Examples: + owi query stats -R lexis:latest/collectionName=cefal + owi --format json query stats -R lexis:latest -o stats.json + owi query stats -R lexis:latest --topk 50 + """ + cli_ctx: CLIContext = ctx.obj + + result = execute_command( + cli_ctx, + query_stats, + manager=cli_ctx.owi, + local_specifier=local, + remote_specifier=remote, + files=files, + topk=topk, + format=cli_ctx.output_format, + output_file=cli_ctx.output_file, + write_back=write_back, + force_write=force_write, + console=cli_ctx.console, + command_name="query stats" + ) + + if result and not result.success: + raise typer.Exit(code=1) + + @app.command() def slice( ctx: typer.Context, diff --git a/owilix/core/tasks/query.py b/owilix/core/tasks/query.py index dd00a24..5a2f354 100644 --- a/owilix/core/tasks/query.py +++ b/owilix/core/tasks/query.py @@ -702,3 +702,205 @@ def query_aggregate( return CommandResult(success=False, msg=str(e)) return CommandResult(success=True, object=df, msg=f"Aggregated {len(_aggregates)} records") + + +def query_stats( + manager: Any, + local_specifier: str, + remote_specifier: str, + files: str = "**/*.parquet", + topk: int = 20, + format: str = "table", + output_file: Optional[str] = None, + write_back: bool = False, + force_write: bool = False, + pq_batch_size: int = 10, + console: Optional[Console] = None +) -> CommandResult: + """ + Calculate statistics over datasets using DuckDB queries. + + Statistics include: + - Record count + - URL suffix distribution (top N) + - URL domain distribution (top N) + - Date range (min/max warc_date) + - MIME type distribution + + Args: + write_back: If True, save stats to .stats folder in dataset + force_write: Force overwrite existing stats files + """ + if console is None: + console = Console() + + from datetime import datetime + import json as json_lib + from collections import Counter + + is_json_mode = format in ["json", "jsonl", "json+", "jsonl+"] + + try: + files_pattern = eval(files) if files is not None else None + except: + files_pattern = [files] + + all_files_by_dataset = get_all_files( + manager, files_pattern, local_specifier, remote_specifier, + console=console, print_it=(format == "table" or output_file) + ) + all_files = group_all_files_by_fs(all_files_by_dataset) + + total_files = sum([len(v) for v in all_files.values()]) + if format == "table" or output_file: + console.print(f"Found '{total_files}' parquet files in {len(all_files)} filesystems. Calculating statistics.") + + if not total_files: + console.print("[yellow]No parquet files found.[/yellow]") + return CommandResult(success=False, msg="No files found") + + stats = { + "generated_at": datetime.utcnow().isoformat() + "Z", + "file_count": total_files, + "record_count": 0, + "url_suffix_distribution": [], + "url_domain_distribution": [], + "date_range": {"min": None, "max": None}, + "mime_type_distribution": [], + } + + # Use aggregate queries through OWIDuckDBSelectExecutor + suffix_counter = Counter() + domain_counter = Counter() + mime_counter = Counter() + min_date = None + max_date = None + record_count = 0 + + # Query for aggregated stats + sql = (OWIlixSQLQuery.from_templates("pq_select") + .select("url_suffix, url_domain, mime_type, warc_date, 1 as cnt") + .where("") + .groupby("") + .partitioned_by("") + .postfix("") + .limit(None)) + + db = OWIDuckDBSelectExecutor(all_files, sql, + pq_batch_size=pq_batch_size, + batch_size=1000, + prefetch=1) + + files_processed = 0 + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + "[progress.percentage]{task.percentage:>3.0f}%", + TextColumn("[cyan]{task.fields[files_done]}/{task.fields[files_total]} files"), + TextColumn("[blue]{task.fields[records]:,} records"), + console=console, + transient=False # Keep visible during processing + ) as progress: + task = progress.add_task( + "Calculating statistics...", + total=total_files, + records=0, + files_done=0, + files_total=total_files + ) + + for results in db.query_aggregator(): + if not results.success: + logger.warning(f"Query error: {results.error}") + continue + + for row in results.rows: + record_count += 1 + + # Extract fields (row is a dict or named tuple) + if isinstance(row, dict): + suffix = row.get("url_suffix") + domain = row.get("url_domain") + mime = row.get("mime_type") + date = row.get("warc_date") + else: + suffix = getattr(row, "url_suffix", None) + domain = getattr(row, "url_domain", None) + mime = getattr(row, "mime_type", None) + date = getattr(row, "warc_date", None) + + if suffix: + suffix_counter[suffix] += 1 + if domain: + domain_counter[domain] += 1 + if mime: + mime_counter[mime] += 1 + if date: + date_str = str(date) + if min_date is None or date_str < min_date: + min_date = date_str + if max_date is None or date_str > max_date: + max_date = date_str + + # Update progress with file batch info + files_processed += len(results.parquet_batch.files) + progress.update( + task, + completed=files_processed, + records=record_count, + files_done=files_processed + ) + + # Build stats from counters + stats["record_count"] = record_count + stats["url_suffix_distribution"] = [{"suffix": k, "count": v} for k, v in suffix_counter.most_common(topk)] + stats["url_domain_distribution"] = [{"domain": k, "count": v} for k, v in domain_counter.most_common(topk)] + stats["mime_type_distribution"] = [{"mime_type": k, "count": v} for k, v in mime_counter.most_common(topk)] + stats["date_range"] = {"min": min_date, "max": max_date} + + # Output results + if is_json_mode: + json_output = json_lib.dumps(stats, indent=2, default=str) + if output_file: + with open(output_file, "w", encoding="utf-8") as f: + f.write(json_output) + console.print(f"[green]Stats written to {output_file}[/green]") + else: + console.print(json_output) + else: + # Rich table output + console.print(f"\n[bold]Dataset Statistics[/bold]") + console.print(f"Generated: {stats['generated_at']}") + console.print(f"Files: {stats['file_count']}") + console.print(f"Records: {stats['record_count']:,}") + + if stats["date_range"]["min"]: + console.print(f"Date Range: {stats['date_range']['min']} to {stats['date_range']['max']}") + + if stats["url_suffix_distribution"]: + table = Table(title="URL Suffix Distribution", show_header=True) + table.add_column("Suffix", style="cyan") + table.add_column("Count", justify="right") + for item in stats["url_suffix_distribution"][:10]: + table.add_row(item["suffix"], f"{item['count']:,}") + console.print(table) + + if stats["url_domain_distribution"]: + table = Table(title="Top Domains", show_header=True) + table.add_column("Domain", style="cyan") + table.add_column("Count", justify="right") + for item in stats["url_domain_distribution"][:10]: + table.add_row(item["domain"], f"{item['count']:,}") + console.print(table) + + if stats["mime_type_distribution"]: + table = Table(title="MIME Types", show_header=True) + table.add_column("MIME Type", style="cyan") + table.add_column("Count", justify="right") + for item in stats["mime_type_distribution"][:10]: + table.add_row(item["mime_type"], f"{item['count']:,}") + console.print(table) + + return CommandResult(success=True, object=stats, msg=f"Calculated stats for {stats['record_count']} records") + + diff --git a/tests/owilix/cli/test_query_less_integration.py b/tests/owilix/cli/test_query_less_integration.py index 84c926e..3ba03bd 100644 --- a/tests/owilix/cli/test_query_less_integration.py +++ b/tests/owilix/cli/test_query_less_integration.py @@ -12,7 +12,7 @@ from pathlib import Path class TestQueryLessIntegration: """Integration tests for owi query less command.""" - REMOTE_SPEC = "all:latest/collectionName=cefal" + REMOTE_SPEC = "all/id=0350fecc-e58b-11f0-a8c9-8ebf6bb2cab9" def run_owi_command(self, args, input_text=None): """Run owi command and return output.""" @@ -220,7 +220,7 @@ class TestQueryLessEdgeCases: result = self.run_owi_command([ "--format", "jsonl", "query", "less", - "--remote", "all:latest/collectionName=cefal", + "--remote", "all/id=0350fecc-e58b-11f0-a8c9-8ebf6bb2cab9", "--where", "url_domain='nonexistent-domain-xyz123'", "--limit", "10" ]) @@ -234,7 +234,7 @@ class TestQueryLessEdgeCases: "--format", "jsonl", "--loglevel", "DEBUG", "query", "less", - "--remote", "all:latest/collectionName=cefal", + "--remote", "all/id=0350fecc-e58b-11f0-a8c9-8ebf6bb2cab9", "--verbose", "--limit", "2" ]) @@ -247,7 +247,7 @@ class TestQueryLessEdgeCases: class TestQuerySitesIntegration: """Integration tests for owi query sites command.""" - REMOTE_SPEC = "all:latest/collectionName=cefal" + REMOTE_SPEC = "all/id=0350fecc-e58b-11f0-a8c9-8ebf6bb2cab9" def run_owi_command(self, args, input_text=None): """Run owi command and return output.""" -- 2.51.2