From eedc6541327069927a2f814b6399a8ddbb59b690 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 26 Oct 2026 18:16:02 -0400 Subject: [PATCH 1/2] feat(asap-tools): Alibaba replay from tarballs and any value column Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GLAs2FV3uZpVD7mUb6vUxk make_replay.py reads MSRTMCR tarballs and takes --column (e.g. providerrpc_mcr); empty values become 0. ALIBABA_MAPPING.md documents the mapping. --- .../datasets/alibaba/ALIBABA_MAPPING.md | 80 +++++++++++++++++++ .../datasets/alibaba/make_replay.py | 26 ++++-- 2 files changed, 99 insertions(+), 7 deletions(-) create mode 100644 asap-tools/experiments/datasets/alibaba/ALIBABA_MAPPING.md diff --git a/asap-tools/experiments/datasets/alibaba/ALIBABA_MAPPING.md b/asap-tools/experiments/datasets/alibaba/ALIBABA_MAPPING.md new file mode 100644 index 00000000..2f8057c1 --- /dev/null +++ b/asap-tools/experiments/datasets/alibaba/ALIBABA_MAPPING.md @@ -0,0 +1,80 @@ +# Alibaba trace → random query workloads (handoff) + +**Goal:** realistic, strong end-to-end results on Alibaba data for the eval section. + +The motivation section (Prometheus, PromSketch, PilotDB; cost vs accuracy) should ideally use the same data, label mapping and similar queries, so the two line up. They don't need to match exactly. + +## Data the end-to-end runs feed + +**Source:** Alibaba cluster-trace-microservices-v2022, MSMetrics, pod `cpu_utilization` (0–1). +- Columns: `timestamp, msname, msinstanceid, nodeid, cpu_utilization, memory_utilization`. +- 60 s step, about 470k pods per step, 28k distinct msname. +- Only about the first hour is local: `benchmarks/metrics_observability/data/alibaba-cluster-data/cluster-trace-microservices-v2022/0d0_0d1/data/MSMetrics/`. + - That is `MSMetricsUpdate_0.csv` plus `MSMetrics_1.tar.gz`. + - Rows are **not** sorted by time. + +**Subset:** fit to our synthetic label shape, so the existing workloads, plans and configs apply unchanged. +- 4 region × 5 service × 10 job × 50 label_0 = 10,000 series. +- `job` = one microservice (msname). + - Seed 862 draws them from the 1,785 msnames with ≥50 pods reporting at all 60 steps. +- `label_0` = one of its pods (msinstanceid; the first 50 by pod number). +- `region` and `service` are synthetic groupings of microservices. + - The trace has neither level. + - `nodeid` cuts across msname, so it doesn't nest. +- Label values: region `r`, service `r.s`, job `r.s.j`, label_0 `r.s.j.k`. + +**Values:** real pod CPU. + +| Statistic | Value | +|---|---| +| min | 0.003 | +| median | 0.12 | +| p95 | 0.40 | +| p99 | 0.58 | +| max | 0.997 | + +- No zeros. +- The busiest microservice is about 5× the median one. +- The top 10 of 200 microservices hold 15% of total CPU. +- Region totals are 20–26k. + +**Time mapping:** a quick-result compromise. +- Replayed at 10 samples/s per series, the same as the synthetic base set. +- One trace minute becomes one 100 ms sample, so the 60-step hour loops every 6 s. +- Spread and skew across pods and services are real; each series' pattern over time repeats. +- A realistic run needs more trace (`fetchData.sh`) and a smaller speedup. + +**Files** (generated, not tracked): + - `replay.csv`: `offset_ms,region,service,job,label_0,value`; 600k rows, sorted by offset. + - `replay_mapping.csv`: our labels → msname/msinstanceid. + - `replay_table.json`: a paced-feeder workload table with `values.replay`. +- Generator: `asap-tools/experiments/datasets/alibaba/make_replay.py`. + +## Query workloads + +Seed 862, 10 random workloads; manifests are in `asap-tools/experiments/datasets/fake_metric/random_workloads/seed-862/`. + +Each end-to-end run uses a workload's 7 `common-promql` queries: +- **Metric and groupings:** metric `data`, grouped by `(region)`, `(region, service)` or `(region, service, job)`. +- **Sums (2):** `sum by (G) (sum_over_time(data[W]))`. + + | Window W | Repeated every | + |---|---| + | 1m | 1–2 s | + | 5m | 5–10 s | + | 10m | 10–20 s | + | 20m | 30–60 s | + +- **Quantiles (3):** 100 s lookback, repeated every 1 s. + - PromQL can't express a pooled quantile, so it uses `avg by (G) (quantile_over_time(phi, data[100s]))`. + - Every system, ClickHouse and LiteOlly included, currently computes this same `avg` of per-series quantiles, so all answers agree. + - The intended pooled `quantile by (G)` over 100 s is being added (LiteOlly support merged in #887–890). +- **Top-k (2):** `topk(k, …)` over counts, with K capped at 1,000 keys. + +On Alibaba data these read naturally: "a quantile of pod CPU per microservice or region over the last 100 s", and "total CPU per service over 5–20 min". + +## End-to-end status (2026-10-15) + +- **Synthetic base:** 5 systems × 10 workloads, all valid. + - LiteOlly vs worst baseline: 6.2× less CPU than raw ClickHouse, and 11.6× lower p95 than Prometheus. +- **Alibaba:** pilot w01 × 5 systems is finishing. The full 10 workloads follow if it is valid. diff --git a/asap-tools/experiments/datasets/alibaba/make_replay.py b/asap-tools/experiments/datasets/alibaba/make_replay.py index 1f5b469a..4f2691ad 100644 --- a/asap-tools/experiments/datasets/alibaba/make_replay.py +++ b/asap-tools/experiments/datasets/alibaba/make_replay.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 -"""A replay table for paced-feeder from Alibaba v2022 MSMetrics: pod CPU -utilization in the random workloads' label shape (4 region x 5 service x 10 +"""A replay table for paced-feeder from Alibaba v2022 per-pod metrics (MSMetrics +CPU by default, or an MSRTMCR call rate with --column) in the random workloads' +label shape (4 region x 5 service x 10 job x 50 label_0), so their plans apply unchanged. Each job is one microservice and each label_0 one of its pods; region and @@ -17,22 +18,32 @@ Usage: python3 make_replay.py MSMetricsUpdate_0.csv MSMetricsUpdate_1.csv \ --out replay [--table eval_table.json] + python3 make_replay.py MSRTMCR_*.tar.gz --column providerrpc_mcr --out calls """ import argparse import csv +import io import json import random +import tarfile from collections import defaultdict REGIONS, SERVICES, JOBS, PODS = 4, 5, 10, 50 def rows(paths): + """The CSV rows of `paths`: plain CSVs, or tarballs of them.""" for path in paths: - with open(path, newline="") as f: - for r in csv.DictReader(f): - yield r + if path.endswith(".tar.gz"): + with tarfile.open(path) as tar: + for member in tar: + if member.isfile(): + text = io.TextIOWrapper(tar.extractfile(member), newline="") + yield from csv.DictReader(text) + else: + with open(path, newline="") as f: + yield from csv.DictReader(f) def main(): @@ -40,6 +51,7 @@ def main(): parser.add_argument("csv", nargs="+") parser.add_argument("--out", required=True) parser.add_argument("--seed", type=int, default=862) + parser.add_argument("--column", default="cpu_utilization", help="the value column") parser.add_argument("--table", help="workload table whose `data` gets the replay") args = parser.parse_args() @@ -72,7 +84,7 @@ def main(): start = min(int(s) for s in steps) out = sorted( - (int(r["timestamp"]) - start, *labels[r["msinstanceid"]], r["cpu_utilization"]) + (int(r["timestamp"]) - start, *labels[r["msinstanceid"]], r[args.column]) for r in rows(args.csv) if r["msinstanceid"] in labels ) @@ -90,7 +102,7 @@ def write_table(table_path, out_path, rows, steps): paced-feeder's series order (mixed radix, label_0 fastest).""" by_series = {} for offset, *_, label_0, value in rows: # sorted by offset - by_series.setdefault(label_0, []).append(round(float(value), 6)) + by_series.setdefault(label_0, []).append(round(float(value or 0), 6)) values = [] for r in range(REGIONS): for s in range(SERVICES): From 63960a935db81964ae1f1761d07178e5f6fbf5c4 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 26 Oct 2026 18:16:18 -0400 Subject: [PATCH 2/2] feat(asap-tools): Google ClusterData 2011 replay builder make_google_replay.py builds a paced-feeder replay table from task_usage CPU rate in the random workloads' label shape. GOOGLE_MAPPING.md and scripts/ record how the trace subset was chosen. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GLAs2FV3uZpVD7mUb6vUxk --- .../datasets/google/GOOGLE_MAPPING.md | 202 +++++++++++++++ .../datasets/google/make_google_replay.py | 190 ++++++++++++++ .../datasets/google/scripts/analyze_subset.py | 231 ++++++++++++++++++ .../datasets/google/scripts/eligible.py | 41 ++++ .../datasets/google/scripts/steps.py | 36 +++ .../datasets/google/scripts/summarize.py | 65 +++++ .../datasets/google/scripts/to_parquet.py | 59 +++++ .../datasets/google/scripts/windows.py | 27 ++ 8 files changed, 851 insertions(+) create mode 100644 asap-tools/experiments/datasets/google/GOOGLE_MAPPING.md create mode 100644 asap-tools/experiments/datasets/google/make_google_replay.py create mode 100644 asap-tools/experiments/datasets/google/scripts/analyze_subset.py create mode 100644 asap-tools/experiments/datasets/google/scripts/eligible.py create mode 100644 asap-tools/experiments/datasets/google/scripts/steps.py create mode 100644 asap-tools/experiments/datasets/google/scripts/summarize.py create mode 100644 asap-tools/experiments/datasets/google/scripts/to_parquet.py create mode 100644 asap-tools/experiments/datasets/google/scripts/windows.py diff --git a/asap-tools/experiments/datasets/google/GOOGLE_MAPPING.md b/asap-tools/experiments/datasets/google/GOOGLE_MAPPING.md new file mode 100644 index 00000000..e3d6a9ba --- /dev/null +++ b/asap-tools/experiments/datasets/google/GOOGLE_MAPPING.md @@ -0,0 +1,202 @@ +# Mapping the Google cluster trace onto the e2e dashboard workload + +All numbers below come from scripts in `scripts/`. Their Parquet intermediates have been deleted; rebuild them with `scripts/to_parquet.py` and then `scripts/steps.py` (about 8 min). + +## 1. Datasets found + +Only one Google dataset is local: **ClusterData 2011-2** (one cell of about 12.5k machines, May 2011), complete, at +`benchmarks/metrics_observability/data/google-cluster-data/ClusterData2011/clusterdata-2011-2/` (42 GB, gzip CSV, no header; columns are in `schema.csv`). + +| table | parts | size | +|---|---|---| +| task_usage | 500 | 40 GB, 1,232,799,308 rows | +| task_events | 500 | 1.6 GB | +| job_events | 500 | 84 MB | +| task_constraints | 500 | 149 MB | +| machine_attributes / machine_events | 1 / 1 | 137 MB / 344 KB | + +- There is no 2019 trace and no Google microservices trace locally. + - `README-google-microservices.md` is a deployment guide for the Online Boutique demo, not a dataset. + - `asap-tools/experiments/datasets/download_google_cluster_2019.sh` fetches 2019 `instance_usage` from GCS on cluster nodes. +- **Time span.** `start_time` runs from 600 s to 2,506,200 s, which is 29.0 days. + - Each part covers about 5,010 s (83.5 min) and the parts are ordered by start time. + - Part `p` covers roughly `[600 + 5011·p, …)` s. +- **Resolution.** Each row is one measurement window, normally 300 s. + - 88% of rows are 300-s windows aligned on both ends. The rest are cut short by task start, stop or migration. + - The step index is `start_time // 300 s`. Every one of the 8,352 steps (2..8353) is present. +- **task_usage columns:** + - keys: start/end time (µs), job ID, task index, machine ID; + - CPU rate (normalized cores); canonical, assigned and max memory; unmapped and total page cache; + - disk I/O time, local disk, max CPU, max disk I/O, CPI, MAI; + - sample portion, aggregation type, sampled CPU. +- **Entity counts:** + - 1 cluster; + - 933 users (925 appear in task_usage); + - 672,074 jobs, of which 654,676 appear in task_usage; + - 39,730 logical job names; + - 24,281,242 tasks, keyed by (job, task index); + - 12,580 machines. +- **Concurrency.** At a typical step, a median of 4,122 jobs and 126,749 tasks are reporting. +- **Duplicates.** Rows per (job, task, step) after bucketing: 1 row for 93.2% of cells, 2–5 rows for 6.4%, and up to more than 30 rows for the rest. These come from restarts and migrations within a step. + +## 2. Hierarchy and mapping + +- **Natural hierarchy:** cluster (1) → user (933) → job (collection) → task index. + - Machine is orthogonal: tasks migrate between machines. + - The schedulable class, scheduling class (0–3), is a per-job attribute. +- **Requirement:** 200 jobs, each with 50 tasks that report at every step of the replayed window. + + | window | strict (every step) | ≥99% of steps, gaps forward-filled | + |---|---|---| + | best single day | 253 (day 13); 224 (day 0) | — | + | best 3 days | **227 (days 26–28)** | — | + | best 4 / 7 days | 197 / 132 | 208 (days 11–17) | + | 14 / 29 days | 41 / 5 | 181 / 142 | + + For the `half` variant (25 tasks per job), 7 days strict gives 215 jobs (days 3–9). +- **Which levels can be real:** + - **job = job ID** and **label_0 = task index** are both real. + - **service = user is not possible.** Among the 227 eligible jobs, only 5 users own 10 or more jobs (one owns 38). We would need 20 such users. + - **region = scheduling class is not possible** either. Class 0 has only about 18 of the 200 chosen jobs; the split is 18/74/53/55, and each region needs 50. + - So **region and service are synthetic**, as for Alibaba. +- **Recommended "user-sorted" grouping.** Order the 200 chosen jobs by (user, job ID), then: + - service = consecutive blocks of 10; + - region = consecutive blocks of 50. + + A big user's jobs then fill whole services, so a service roughly stands for an owner or team. This keeps real skew at the service level: service-level Zipf is 0.75 against 0.52 for a random grouping (CPU). The Alibaba-style random grouping also works but separates the (region, service) top-k less well (see §3). + +## 3. Measurements + +The subset is days 26–28, strict coverage, `random.Random(862).sample(sorted eligible job IDs, 200)`, first 50 tasks by index. That is 10,000 series × 864 steps with no missing cells. + +### Value distribution (per-step, deduplicated values) + +| metric | min | p50 | p95 | p99 | max | zeros | < 5e-5 (→ 0 at ×1e4) | min > 0 | +|---|---|---|---|---|---|---|---|---| +| CPU rate | 0 | 0.00472 | 0.0906 | 0.178 | 0.640 | 1.5% | 6.0% (pool) | 9.5e-7 (= 2⁻²⁰) | +| canonical mem | 0 | 0.00633 | 0.0797 | 0.168 | 0.227 | 1.4% | 1.5% | 9.5e-7 | + +- **Value format.** Values are fractions (normalized to the largest machine). They are floats rounded to about 4 significant digits; the smallest ones are multiples of 2⁻²⁰. None are integers. +- **Variation over time.** Median per-series CV over time is 0.29 for CPU and 0.034 for memory. Memory is almost flat in time. +- **Dead series.** 91 of the 10,000 tasks have zero CPU over all 3 days, and 1 chosen job is entirely zero. +- **Outliers.** CPU above 1 is rare (1.5e-6 of raw rows; raw max 3.96, and 5.23 after per-step averaging on day 13). There are none in days 26–28. + +### Zipf MLE of totals (`zipf_mle` copied from `fit_skew.py`), CPU / mem + +| level | CPU | mem | +|---|---|---| +| leaf (10,000 tasks) | 0.68 | 0.64 | +| job (200 chosen) | 0.86 | 0.80 | +| job (all 227 eligible) | 1.08 | 1.08 | +| service (20), user-sorted / random | 0.75 / 0.52 | 0.61 / 0.50 | +| region (4), user-sorted / random | 0.51 / 0.53 | 0.43 / 0.49 | +| whole cluster, by job_id / by user (from `skew_summary.csv`) | 1.10 / 1.19 | — | + +For comparison, Alibaba CPU had Zipf 0.49 by microservice and 0.43 by pod. Google is clearly more skewed at the job level. + +Within a job, tasks are near-ties, just as Alibaba's pods were: +- median CV of task totals within a job: 0.083; +- median gap between the top two tasks: 2.1%; +- median ratio of max to median task: 1.17. + +So top-k must rank aggregates, never tasks. + +### Top-k separability, CPU + +Each cell shows top-k share / relative gap (v_k − v_{k+1}) / v_k. + +- "full" is the totals over the whole 864-step loop. +- "1m" means every cyclic 600-sample window, i.e. `sum_over_time(data[1m])`. Each cell gives the median gap, the minimum gap, and the fraction of windows whose top-k set equals the full-loop top-k. +- The 5m and 20m windows are about 3.5 and 14 whole loops, so they look like "full", with minimum gaps ≥ 0.019 everywhere for user-sorted grouping and for job-level top-k. + +| grouping / level | k=3 | k=5 | k=8 | +|---|---|---|---| +| **user-sorted (region, service)** full | 0.36 / 0.050 | 0.55 / 0.149 | 0.73 / 0.110 | +| user-sorted (region, service) 1m | gap 0.050, min 0.027, same set 100% | 0.148, min 0.069, 100% | 0.110, min 0.085, 100% | +| random (region, service) full | 0.28 / 0.068 | 0.43 / 0.062 | 0.62 / 0.030 | +| random (region, service) 1m | 0.071, min 0.044, 100% | 0.057, min 0.045, 100% | 0.024, **min 0.000, 92%** | +| (region, service, job) full | 0.16 / 0.269 | 0.22 / 0.022 | 0.31 / 0.022 | +| (region, service, job) 1m | 0.270, min 0.254, 100% | 0.020, **min 0.000, 83%** | 0.015, **min 0.0002, 88%** | + +Memory gaps are similar in size, but memory has a tie at the (region, service) k=5 boundary under random grouping (gap 0.000 on day 13 and 0.020 on days 26–28). + +Other windows tried (full tables in `scratch/summary.txt`): +- **Days 11–17, ≥99% coverage** (2,016 steps): there is more temporal churn. At 1m, the job-level k=8 set never matches the full-loop set, and the random (region, service) k=3 matches in only 75% of windows. +- **Day 13** (288 steps): a 1m window is 2.1 loops, so rankings are almost static. Its (region, service) gaps are large (0.09–0.29). + +## 4. Time mapping + +- **Step and loop length.** The step is 300 s, versus 60 s for Alibaba. With one step per sample at 10 samples/s: + + | window | steps | loop length | + |---|---|---| + | 1 day | 288 | 28.8 s | + | **3 days (recommended)** | 864 | 86.4 s | + | 7 days | 2,016 | 201.6 s | + | whole trace | 8,352 | 835 s ≈ 13.9 min | + + Alibaba's hour looped every 6 s. +- **How much trace a query window covers.** `[1m]` is 600 steps (50 h of trace), `[5m]` is 10.4 days, and `[20m]` is 41.7 days. Every lookback of 5m or more wraps the loop at any window length the data supports. +- **Fetching more is not worth it for 2011.** The full trace is already local; the limit is task churn, not data. Longer windows lose the "50 tasks at every step" jobs (4 days: 197; 29 days: 5). +- **The 2019 trace (8 cells, 31 days, 5-min `instance_usage`) is the only upgrade.** + - Gain: **region = cell** becomes real, and collection → instance replaces job → task. + - Cost: tens to hundreds of GB of `instance_usage` per cell via gsutil or BigQuery. + - It would not fix the 5-min step or the churn. Worth it only if a real top level matters for the paper. + +## 5. Recommendation + +- **Metric: CPU rate** (task_usage column 6). + - It varies over time (CV 0.29) and has a wide value spread (p99/p50 ≈ 38×), which suits the quantile queries. + - Its job-level skew is 0.86 versus 0.49 for Alibaba. + - Canonical memory has similar skew but is nearly constant in time. +- **Mapping:** + - job = job ID, label_0 = task index (the first 50 by index); + - region and service are synthetic, as blocks over the jobs sorted by (user, job); + - window = days 26–28 (steps 7490..8353, `start_time` in [2,247,000, 2,506,200) s, parts 00448–00499); + - strict coverage, 227 eligible jobs → 200 chosen with seed 862. + - The fallback that needs no extra input is Alibaba-style random grouping, but its (region, service) k=8 gap is thinner. +- **Scaling: ×1,000,000.** Counters in the Count-Min-with-heap sketch are i64 and the update does `value.round() as i64` (asap_sketchlib rev 5158bf2, `countminsketch_topk.rs`). + - At ×10,000, 6.0% of samples round to 0, and 4.3 points of that are real nonzero values. + - At ×1e6 the smallest nonzero value (9.5e-7) becomes 1 and no nonzero sample is lost. + - The largest sum is about 0.64e6 × 12,000 samples × 500 series ≈ 4e12, well inside i64. + - If you'd rather reuse `synthetic_data_value_scale = 10000` unchanged, expect about 6% zeros. +- **Changes `make_replay.py` needs** (a `datasets/google/make_replay.py`): + 1. **Input.** Read headerless `task_usage/part-00448..00499-of-00500.csv.gz` by column position: + - 0 = start µs, 1 = end µs, 2 = job ID, 3 = task index, 5 = CPU rate. + - Also read `job_events` (col 2 = job ID, col 4 = user) for the grouping; take the last non-null value per job. + 2. **Steps.** `step = start_us // 300_000_000`. Keep 7490 ≤ step < 8354. + 3. **Series key.** Use `(job, task index)`. Do not use machine, because tasks migrate between machines. + 4. **Dedup.** For each (job, task, step), take the duration-weighted mean of CPU rate with weight `max(end − start, 1 s)`. About 7% of cells have more than one row. + 5. **Eligibility.** A task is eligible if it reports at all 864 steps. A job is eligible if it has at least 50 such tasks. + - Sort job IDs as ints, then `random.Random(862).sample(eligible, 200)` gives exactly the subset measured here. + - Optionally drop tasks whose CPU is zero at every step before taking the first 50. + 6. **Labels.** Order the chosen jobs by (user, job). Use `region = i // 50`, `service = f"{region}.{i // 10 % 5}"`, `job = f"{service}.{i % 10}"`, and `label_0 = f"{job}.{k}"`. + 7. **Mapping CSV.** Write columns `region, service, job, label_0, user, job_id, task_index`. + 8. **Offsets.** `offset_ms = (step − 7490) × 300,000`. The value is CPU × 1e6. + 9. **Table.** `write_table` is unchanged: 864 steps, series-major. + 10. **Implementation.** Don't hold per-pod timestamp sets the way the Alibaba script does. The window has 116M rows and 2.85M tasks, which would take more than 10 GB in Python sets. + - Aggregate with DuckDB instead. It is in `baselines/PilotDB/.venv`. + - Or start from `scratch/steps.parquet`, which is already deduplicated per (job, task, step) with CPU and memory. + - `scripts/analyze_subset.py` already contains the selection logic. +- **Risks:** + - **Near-ties at the boundary.** + - Job-level k=5 and k=8 (gaps 2.2%) reach a gap of 0 in some 1m windows, and their top-k sets match the full-loop sets in only 83% and 88% of windows. + - Under random grouping, (region, service) k=8 does the same (92%). + - User-sorted grouping removes the (region, service) ties. Job-level ties come from the data, so evaluate top-k with gap-aware or set-overlap metrics. + - **Task-level flatness.** Leaf-level Zipf is 0.68 and tasks within a job are near-ties. That is fine for the pooled queries, which never rank tasks. + - **Zeros and sparse series.** 1.5% of samples are 0, 5% of series contain at least one 0, and 91 tasks (1 whole job) are all-zero. Coverage is complete with no gaps in the strict window. The 7-day ≥99% variant forward-fills 0.09% of cells. + - **Short loop relative to the lookbacks.** 5m and 20m windows sum about 3.5 and 14 identical loops, so they rank like the full loop, and only 1m (0.69 loop) sees temporal churn. + - **Outliers.** CPU above 1.0 exists elsewhere in the trace (day 13: 5.23). Clip at 1.0 if you use another window. + - **Selection bias.** Strict coverage favors long-running service jobs. That is realistic for dashboards but excludes batch churn. + +## 6. Built replay + +`make_google_replay.py` implements §5. It reads the raw CSVs with DuckDB, needs no scratch files, and takes about 40 s with 2.9 GB peak RSS. The output is byte-identical across thread counts. + +- **Outputs:** + - `replay_table_google.json` (70 MB): 10,000 series × 864 steps, no NaNs. `paced_feeder --verify-facts` passes. + - `replay_mapping.csv` + - `replay.csv` (332 MB) +- **Values (CPU × 1e6):** p50 4,715; p95 90,580; p99 178,200; max 639,600; 1.46% zeros. 91 series are zero at every step. +- **Fractional values.** 28% of values are fractional: these are the raw values below 0.001, which have 4 significant digits. Count-Min rounds them. +- **Skew.** Zipf is 0.861 for jobs, 0.750 for services and 0.676 for tasks. Top-k figures match §3. diff --git a/asap-tools/experiments/datasets/google/make_google_replay.py b/asap-tools/experiments/datasets/google/make_google_replay.py new file mode 100644 index 00000000..952deed1 --- /dev/null +++ b/asap-tools/experiments/datasets/google/make_google_replay.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python3 +"""A replay table for paced-feeder from Google ClusterData 2011-2 task_usage: +task CPU rate in the random workloads' label shape (4 region x 5 service x 10 +job x 50 label_0), so their plans apply unchanged. + +The Google counterpart of asap-tools/experiments/datasets/alibaba/make_replay.py, +with the same outputs and label convention (a child's value is its parent's, a +`.`, and its index), and these differences: +- input is the headerless task_usage CSV.gz parts (read with DuckDB) plus + job_events for each job's user; a step is `start_time // 300 s`; +- a series is a Google task (job ID, task index), not a machine; rows of one + task in one step (restarts, migrations) are merged by a duration-weighted + mean of CPU rate; +- job = Google job (drawn by `--seed` among jobs with >= 50 tasks reporting at + every step, sorted as integers; tasks are their first 50 by index); +- region/service are blocks of 50/10 of the chosen jobs ordered by (user, job), + so one user's jobs share services; +- values are CPU rate clipped at 1.0, times `--scale` (1e6), rounded to 2 + decimals: Count-Min top-k counters round to integers, and x1e4 would zero + about 6% of samples. + +Writes `/replay_table_google.json` (the workload table with `data`'s +values replaced by `{"replay": {"steps", "values"}}`, series-major in +paced-feeder's series order: mixed radix over region, service, job, label_0), +`/replay_mapping.csv`, and `/replay.csv` +(`offset_ms,region,service,job,label_0,value`, sorted by offset). + +Usage (DuckDB from baselines/PilotDB/.venv): + python -I make_google_replay.py --trace .../clusterdata-2011-2 \ + --table .../seed-862-dashboard/workload-01-pilot/eval_table.json --out DIR +""" + +import argparse +import csv +import json +import os +import random +import tempfile + +import duckdb +import numpy as np + +REGIONS, SERVICES, JOBS, PODS = 4, 5, 10, 50 +STEP_US = 300_000_000 +FIRST_STEP, STEPS = 7490, 864 # trace days 26..28 +FIRST_PART, LAST_PART = 448, 499 # task_usage parts covering those steps + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--trace", required=True, help="clusterdata-2011-2 directory") + parser.add_argument( + "--table", required=True, help="workload table whose `data` gets the replay" + ) + parser.add_argument("--out", required=True) + parser.add_argument("--seed", type=int, default=862) + parser.add_argument("--scale", type=float, default=1e6) + parser.add_argument("--memory-limit", default="2GB") + parser.add_argument("--threads", type=int, default=8) + args = parser.parse_args() + os.makedirs(args.out, exist_ok=True) + + con = duckdb.connect() + tmp = tempfile.mkdtemp(dir=args.out, prefix="duckdb-tmp-") + con.execute( + f"SET memory_limit='{args.memory_limit}'; SET threads={args.threads}; SET temp_directory='{tmp}'; " + "SET preserve_insertion_order=false; SET enable_progress_bar=false" + ) + parts = [ + f"{args.trace}/task_usage/part-{p:05d}-of-00500.csv.gz" + for p in range(FIRST_PART, LAST_PART + 1) + ] + lo, hi = FIRST_STEP * STEP_US, (FIRST_STEP + STEPS) * STEP_US + con.execute( + f""" + CREATE TABLE cells AS + SELECT job, task, (start_us // {STEP_US} - {FIRST_STEP})::SMALLINT AS t, + -- exact DECIMAL sums keep the mean independent of thread order + sum(cpu::DECIMAL(18, 12) * w)::DOUBLE / sum(w)::DOUBLE AS cpu + FROM (SELECT column0 AS start_us, column2 AS job, column3 AS task, column5 AS cpu, + greatest(column1 - column0, 1000000) AS w + FROM read_csv({parts!r}, header=false, auto_detect=false, + columns={{'column0':'BIGINT','column1':'BIGINT','column2':'BIGINT','column3':'INTEGER', + 'column4':'BIGINT','column5':'DOUBLE','column6':'VARCHAR','column7':'VARCHAR', + 'column8':'VARCHAR','column9':'VARCHAR','column10':'VARCHAR','column11':'VARCHAR', + 'column12':'VARCHAR','column13':'VARCHAR','column14':'VARCHAR','column15':'VARCHAR', + 'column16':'VARCHAR','column17':'VARCHAR','column18':'VARCHAR','column19':'VARCHAR'}})) + WHERE start_us >= {lo} AND start_us < {hi} + GROUP BY ALL""" + ) + con.execute( + f"CREATE TABLE full_tasks AS SELECT job, task FROM cells GROUP BY ALL HAVING count(*) = {STEPS}" + ) + eligible = [ + j + for (j,) in con.execute( + f"SELECT job FROM full_tasks GROUP BY job HAVING count(*) >= {PODS} ORDER BY job" + ).fetchall() + ] + jobs = REGIONS * SERVICES * JOBS + print(f"{STEPS} steps; {len(eligible)} jobs eligible") + chosen = random.Random(args.seed).sample(eligible, jobs) + + users = dict( + con.execute( + f""" + SELECT column2, arg_max(column4, column0) FILTER (WHERE column4 IS NOT NULL) + FROM read_csv('{args.trace}/job_events/part-*-of-00500.csv.gz', header=false, auto_detect=false, + columns={{'column0':'BIGINT','column1':'VARCHAR','column2':'BIGINT','column3':'INTEGER', + 'column4':'VARCHAR','column5':'VARCHAR','column6':'VARCHAR','column7':'VARCHAR'}}) + WHERE column2 IN (SELECT unnest($chosen)) GROUP BY 1""", + {"chosen": chosen}, + ).fetchall() + ) + ordered = sorted(chosen, key=lambda j: (users[j], j)) + + # series index in paced-feeder's order: ((i_job * PODS) + k), i_job = r*50 + s*10 + j + mapping = [] # (region, service, job, label_0, job_id, task, user, series) + for i, job_id in enumerate(ordered): + region = str(i // (SERVICES * JOBS)) + service = f"{region}.{i // JOBS % SERVICES}" + job = f"{service}.{i % JOBS}" + tasks = [ + t + for (t,) in con.execute( + f"SELECT task FROM full_tasks WHERE job = ? ORDER BY task LIMIT {PODS}", + [job_id], + ).fetchall() + ] + for k, task in enumerate(tasks): + mapping.append( + ( + region, + service, + job, + f"{job}.{k}", + job_id, + task, + users[job_id], + i * PODS + k, + ) + ) + with open(f"{args.out}/replay_mapping.csv", "w", newline="") as f: + w = csv.writer(f) + w.writerow( + [ + "region", + "service", + "job", + "label_0", + "google_job_id", + "task_index", + "user", + ] + ) + w.writerows(m[:7] for m in mapping) + + con.execute("CREATE TABLE sel (job BIGINT, task INTEGER, series INTEGER)") + con.executemany( + "INSERT INTO sel VALUES (?, ?, ?)", [(m[4], m[5], m[7]) for m in mapping] + ) + got = con.execute( + f"""SELECT series, t, round(least(cpu, 1.0) * {args.scale}, 2) AS v + FROM cells JOIN sel USING (job, task)""" + ).fetchnumpy() + n = len(mapping) + values = np.full((n, STEPS), np.nan) + values[got["series"], got["t"]] = got["v"] + assert n == REGIONS * SERVICES * JOBS * PODS and not np.isnan(values).any() + + labels = {m[7]: m[:4] for m in mapping} + with open(f"{args.out}/replay.csv", "w", newline="") as f: + w = csv.writer(f) + w.writerow(["offset_ms", "region", "service", "job", "label_0", "value"]) + for t in range(STEPS): + for s in range(n): + w.writerow([t * STEP_US // 1000, *labels[s], values[s, t]]) + + table = json.load(open(args.table)) + data = table["workloads"][0]["e2e"]["metrics"]["data"] + data["values"] = {"replay": {"steps": STEPS, "values": values.ravel().tolist()}} + with open(f"{args.out}/replay_table_google.json", "w") as f: + json.dump(table, f) + con.close() + os.rmdir(tmp) if not os.listdir(tmp) else None + print(f"{n} series x {STEPS} steps -> {args.out}/replay_table_google.json") + + +if __name__ == "__main__": + main() diff --git a/asap-tools/experiments/datasets/google/scripts/analyze_subset.py b/asap-tools/experiments/datasets/google/scripts/analyze_subset.py new file mode 100644 index 00000000..73ceb03d --- /dev/null +++ b/asap-tools/experiments/datasets/google/scripts/analyze_subset.py @@ -0,0 +1,231 @@ +"""Select an Alibaba-style subset (200 jobs x P tasks) from the deduped Google +2011 steps over a day window and measure value spread, Zipf skew and top-k +separability under two region/service groupings. + +usage: analyze_subset.py SCRATCH DAY0 NDAYS THRESH P [--seed 862] +""" + +import argparse +import json +import random +import duckdb +import numpy as np +from scipy.optimize import minimize_scalar + +ZIPF_THETA_BOUNDS = (0.0, 5.0) + + +def zipf_mle(weights): # copied from asap-tools/dataset-analysis/fit_skew.py + w = np.sort(np.asarray(weights, dtype=float))[::-1] + w = w[w > 0] + if len(w) < 2: + return float("nan") + log_rank = np.log(np.arange(1, len(w) + 1)) + total, weighted_log_rank = w.sum(), (w * log_rank).sum() + + def nll(s): + return s * weighted_log_rank + total * np.log(np.exp(-s * log_rank).sum()) + + return float(minimize_scalar(nll, bounds=ZIPF_THETA_BOUNDS, method="bounded").x) + + +def topk_stats(totals, k): + v = np.sort(totals)[::-1] + return v[:k].sum() / v.sum(), (v[k - 1] - v[k]) / v[k - 1] + + +def cyclic_window_sums(m, w): + """m: [groups, T]; sums over every cyclic window of w steps -> [groups, T].""" + T = m.shape[1] + reps, rem = divmod(w, T) + ext = np.concatenate([m, m[:, :rem]], axis=1) if rem else m + cs = np.concatenate([np.zeros((m.shape[0], 1)), np.cumsum(ext, axis=1)], axis=1) + part = cs[:, rem : rem + T] - cs[:, :T] if rem else 0.0 + return reps * m.sum(axis=1, keepdims=True) + part + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("scratch") + ap.add_argument("day0", type=int) + ap.add_argument("ndays", type=int) + ap.add_argument("thresh", type=float) + ap.add_argument("P", type=int) + ap.add_argument("--seed", type=int, default=862) + ap.add_argument("--out") + a = ap.parse_args() + D = a.scratch + s0, T = 2 + 288 * a.day0, 288 * a.ndays + need = T if a.thresh >= 1 else int(a.thresh * T) + c = duckdb.connect() + c.execute("SET memory_limit='5GB'; SET threads=12") + c.execute( + f"""CREATE TABLE cov AS SELECT job, task, count(*) n FROM '{D}/steps.parquet' + WHERE step >= {s0} AND step < {s0 + T} GROUP BY ALL HAVING count(*) >= {need}""" + ) + elig = [ + r[0] + for r in c.sql( + f"SELECT job FROM cov GROUP BY job HAVING count(*) >= {a.P} ORDER BY job" + ).fetchall() + ] + chosen = random.Random(a.seed).sample(elig, 200) + c.execute( + "CREATE TABLE ch AS SELECT * FROM (SELECT unnest($1::BIGINT[]) job, generate_subscripts($1::BIGINT[], 1) - 1 AS idx)", + [chosen], + ) + c.execute( + f"""CREATE TABLE sel AS SELECT job, task, idx, k FROM (SELECT cov.job, task, idx, + row_number() OVER (PARTITION BY cov.job ORDER BY task) - 1 AS k FROM cov JOIN ch USING (job)) WHERE k < {a.P}""" + ) + users = dict( + c.sql( + f"SELECT job, user FROM '{D}/jobs.parquet' JOIN ch USING (job)" + ).fetchall() + ) + sclass = dict( + c.sql( + f"SELECT job, sclass FROM '{D}/jobs.parquet' JOIN ch USING (job)" + ).fetchall() + ) + rows = c.sql( + f"""SELECT idx, k, step - {s0} AS t, cpu, mem FROM '{D}/steps.parquet' JOIN sel USING (job, task) + WHERE step >= {s0} AND step < {s0 + T}""" + ).fetchnumpy() + N = 200 * a.P + leaf = rows["idx"].astype(np.int64) * a.P + rows["k"] + out = { + "window": f"days {a.day0}..{a.day0 + a.ndays - 1} (steps {s0}..{s0 + T - 1})", + "T": T, + "thresh": a.thresh, + "P": a.P, + "eligible_jobs": len(elig), + "users_among_chosen": len(set(users.values())), + "max_chosen_jobs_per_user": max( + np.unique(list(users.values()), return_counts=True)[1].tolist() + ), + "users_with_ge10_chosen": int( + (np.unique(list(users.values()), return_counts=True)[1] >= 10).sum() + ), + "sclass_counts": { + str(k): v + for k, v in zip(*np.unique(list(sclass.values()), return_counts=True)) + }, + } + for metric in ("cpu", "mem"): + m = np.full((N, T), np.nan) + m[leaf, rows["t"]] = rows[metric] + missing = np.isnan(m).mean() + # forward-fill then back-fill missing steps (only when thresh < 1) + for i in np.where(np.isnan(m).any(axis=1))[0]: + r = m[i] + idx = np.where(~np.isnan(r), np.arange(T), 0) + np.maximum.accumulate(idx, out=idx) + r[:] = r[idx] + first = np.argmax(~np.isnan(r)) + r[np.isnan(r)] = r[first] + v = m.ravel() + res = { + "missing_frac": missing, + "min": v.min(), + "p50": np.median(v), + "p95": np.quantile(v, 0.95), + "p99": np.quantile(v, 0.99), + "max": v.max(), + "zero_frac": (v == 0).mean(), + "lt_1e-4_frac": (v < 1e-4).mean(), + "lt_1e-6_frac": (v < 1e-6).mean(), + "distinct_values": int(np.unique(v.astype(np.float32)).size), + "series_constant_frac": (m.max(axis=1) == m.min(axis=1)).mean(), + "median_series_cv": float( + np.median(m.std(axis=1) / np.maximum(m.mean(axis=1), 1e-12)) + ), + } + leaf_tot = m.sum(axis=1) + job_m = m.reshape(200, a.P, T).sum(axis=1) + within = m.reshape(200, a.P, T).sum(axis=2) + with np.errstate(invalid="ignore", divide="ignore"): + res["within_job_cv_median"] = float( + np.nanmedian(within.std(axis=1) / within.mean(axis=1)) + ) + res["zero_jobs"] = int((within.sum(axis=1) == 0).sum()) + res["zero_tasks"] = int((leaf_tot == 0).sum()) + res["min_nonzero"] = float(v[v > 0].min()) + res["frac_series_with_any_zero"] = float((m == 0).any(axis=1).mean()) + srt = -np.sort(-within, axis=1) # per-job task totals, desc + res["within_job_top1_vs_2_gap_median"] = float( + np.nanmedian( + (srt[:, 0] - srt[:, 1]) / np.where(srt[:, 0] > 0, srt[:, 0], np.nan) + ) + ) + res["within_job_max_over_median_median"] = float( + np.nanmedian( + srt[:, 0] + / np.where(np.median(srt, axis=1) > 0, np.median(srt, axis=1), np.nan) + ) + ) + ej = c.sql( + f"""SELECT sum({metric}) FROM '{D}/steps.parquet' JOIN (SELECT job, task FROM cov WHERE job IN (SELECT unnest($1::BIGINT[]))) USING (job, task) + WHERE step >= {s0} AND step < {s0 + T} GROUP BY job""", + params=[elig], + ).fetchnumpy() + res["zipf_job_eligible_pool_all_full_tasks"] = zipf_mle(list(ej.values())[0]) + res["zipf_leaf"] = zipf_mle(leaf_tot) + res["zipf_job"] = zipf_mle(job_m.sum(axis=1)) + groupings = {"random": np.arange(200)} + order = sorted(range(200), key=lambda i: (users[chosen[i]], chosen[i])) + pos = np.empty(200, dtype=int) + pos[order] = np.arange(200) + groupings["user_sorted"] = pos + for gname, slot in groupings.items(): + # slot[i] = position 0..199 of chosen job i; region = pos//50, service = pos//10 + jm = np.zeros_like(job_m) + jm[slot] = job_m + sm = jm.reshape(20, 10, T).sum(axis=1) + rm = jm.reshape(4, 50, T).sum(axis=1) + g = { + "zipf_service": zipf_mle(sm.sum(axis=1)), + "zipf_region": zipf_mle(rm.sum(axis=1)), + } + for lvl, mat in (("rs", sm), ("rsj", jm)): + for wname, w in ( + ("full", None), + ("1m=600", 600), + ("5m=3000", 3000), + ("20m=12000", 12000), + ): + sums = ( + mat.sum(axis=1, keepdims=True) + if w is None + else cyclic_window_sums(mat, w) + ) + full_top = { + k: set(np.argsort(-mat.sum(axis=1))[:k]) for k in (3, 5, 8) + } + for k in (3, 5, 8): + st = np.array( + [topk_stats(sums[:, t], k) for t in range(sums.shape[1])] + ) + same = np.mean( + [ + set(np.argsort(-sums[:, t])[:k]) == full_top[k] + for t in range(sums.shape[1]) + ] + ) + g[f"{lvl}/{wname}/k{k}"] = { + "share_med": float(np.median(st[:, 0])), + "gap_med": float(np.median(st[:, 1])), + "gap_min": float(st[:, 1].min()), + "gap_p10": float(np.quantile(st[:, 1], 0.1)), + "set_eq_full_frac": float(same), + } + res[gname] = g + out[metric] = res + js = json.dumps(out, indent=1, default=float) + print(js) + if a.out: + open(a.out, "w").write(js) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/experiments/datasets/google/scripts/eligible.py b/asap-tools/experiments/datasets/google/scripts/eligible.py new file mode 100644 index 00000000..4e0a68b2 --- /dev/null +++ b/asap-tools/experiments/datasets/google/scripts/eligible.py @@ -0,0 +1,41 @@ +import sys +import duckdb + +D = sys.argv[1] +con = duckdb.connect() +con.execute(f"SET memory_limit='5GB'; SET threads=12; SET temp_directory='{D}/tmp'") +W = {"h1": (2, 14), "d1": (2, 290), "d7": (2, 2018), "d29": (2, 8354)} +aggs = ", ".join( + f"count(*) FILTER (WHERE step >= {a} AND step < {b}) AS n_{k}" + for k, (a, b) in W.items() +) +con.execute( + f"COPY (SELECT job, task, {aggs} FROM '{D}/steps.parquet' GROUP BY ALL) TO '{D}/task_cov.parquet' (FORMAT parquet)" +) +full = ", ".join( + f"count(*) FILTER (WHERE n_{k} = {b - a}) AS full_{k}" for k, (a, b) in W.items() +) +con.execute( + f"COPY (SELECT job, count(*) AS ntasks, {full} FROM '{D}/task_cov.parquet' GROUP BY job) TO '{D}/job_cov.parquet' (FORMAT parquet)" +) +print( + con.sql( + f"SELECT count(*) tasks, count(DISTINCT job) jobs FROM '{D}/task_cov.parquet'" + ).fetchall() +) +for k in W: + print( + k, + con.sql( + f"""SELECT count(*) FILTER (WHERE full_{k} >= 1) AS jobs_any, count(*) FILTER (WHERE full_{k} >= 50) AS jobs50, + count(DISTINCT j.user) FILTER (WHERE full_{k} >= 50) users50, sum(full_{k}) full_tasks + FROM '{D}/job_cov.parquet' c JOIN '{D}/jobs.parquet' j USING (job)""" + ).fetchall(), + ) + print( + " users by #jobs50:", + con.sql( + f"""SELECT n, count(*) FROM (SELECT user, count(*) n FROM '{D}/job_cov.parquet' c JOIN '{D}/jobs.parquet' j USING (job) + WHERE full_{k} >= 50 GROUP BY user) GROUP BY n ORDER BY n""" + ).fetchall(), + ) diff --git a/asap-tools/experiments/datasets/google/scripts/steps.py b/asap-tools/experiments/datasets/google/scripts/steps.py new file mode 100644 index 00000000..ab2cc8df --- /dev/null +++ b/asap-tools/experiments/datasets/google/scripts/steps.py @@ -0,0 +1,36 @@ +"""task_usage.parquet -> steps.parquet: one row per (job, task, 5-min step), +value = duration-weighted mean over the step's measurement rows (dedups +restarts / split windows). Also per-task coverage stats.""" + +import sys +import time +import duckdb + +D = sys.argv[1] +con = duckdb.connect() +con.execute( + f"SET memory_limit='5GB'; SET threads=12; SET preserve_insertion_order=false; SET temp_directory='{D}/tmp'" +) +t = time.time() +con.execute( + f""" +COPY (SELECT job, task, (start_s // 300)::INTEGER AS step, + (sum(cpu * greatest(end_s - start_s, 1)) / sum(greatest(end_s - start_s, 1)))::FLOAT AS cpu, + (sum(mem * greatest(end_s - start_s, 1)) / sum(greatest(end_s - start_s, 1)))::FLOAT AS mem, + count(*)::SMALLINT AS nrows + FROM '{D}/task_usage.parquet' GROUP BY ALL) +TO '{D}/steps.parquet' (FORMAT parquet, COMPRESSION zstd, ROW_GROUP_SIZE 1000000)""" +) +print("steps done", time.time() - t, flush=True) +print( + con.sql( + f"SELECT count(*), min(step), max(step), count(DISTINCT step) FROM '{D}/steps.parquet'" + ).fetchall(), + flush=True, +) +print( + con.sql( + f"SELECT nrows, count(*) FROM '{D}/steps.parquet' GROUP BY 1 ORDER BY 1 LIMIT 8" + ).fetchall(), + flush=True, +) diff --git a/asap-tools/experiments/datasets/google/scripts/summarize.py b/asap-tools/experiments/datasets/google/scripts/summarize.py new file mode 100644 index 00000000..9c05e16d --- /dev/null +++ b/asap-tools/experiments/datasets/google/scripts/summarize.py @@ -0,0 +1,65 @@ +import json +import sys + +for f in sys.argv[1:]: + d = json.load(open(f)) + print( + "#####", + f.rsplit("/", 1)[1], + d["window"], + "P", + d["P"], + "thresh", + d["thresh"], + "eligible", + d["eligible_jobs"], + "users", + d["users_among_chosen"], + "maxjobs/user", + d["max_chosen_jobs_per_user"], + "users>=10", + d["users_with_ge10_chosen"], + d["sclass_counts"], + ) + for m in ("cpu", "mem"): + r = d[m] + print( + " ", + m, + { + k: (float("%.4g" % v) if isinstance(v, float) else v) + for k, v in r.items() + if not isinstance(v, dict) + }, + ) + for g in ("random", "user_sorted"): + print( + " ", + g, + "zipf svc %.3f region %.3f" + % (r[g]["zipf_service"], r[g]["zipf_region"]), + ) + for lvl in ("rs", "rsj"): + if g == "user_sorted" and lvl == "rsj": + continue + for w in ("full", "1m=600", "5m=3000"): + print( + " %-8s" % (lvl + "/" + w), + " ".join( + "k%d share %.3f gap %.3f (p10 %.3f, min %.3f, set==%.2f)" + % ( + k, + *( + r[g][f"{lvl}/{w}/k{k}"][x] + for x in ( + "share_med", + "gap_med", + "gap_p10", + "gap_min", + "set_eq_full_frac", + ) + ), + ) + for k in (3, 5, 8) + ), + ) diff --git a/asap-tools/experiments/datasets/google/scripts/to_parquet.py b/asap-tools/experiments/datasets/google/scripts/to_parquet.py new file mode 100644 index 00000000..bdee62cc --- /dev/null +++ b/asap-tools/experiments/datasets/google/scripts/to_parquet.py @@ -0,0 +1,59 @@ +"""task_usage CSV.gz -> parquet (needed columns only), plus job_events -> jobs.parquet. + +Usage: python3 to_parquet.py .../clusterdata-2011-2 OUT_DIR""" + +import sys +import time +import duckdb + +G, OUT = sys.argv[1], sys.argv[2] +con = duckdb.connect() +con.execute( + "SET memory_limit='5GB'; SET threads=12; SET preserve_insertion_order=false" +) +con.execute(f"SET temp_directory='{OUT}/tmp'") +cols = ", ".join( + f"'c{i}': '{t}'" + for i, t in enumerate( + [ + "BIGINT", + "BIGINT", + "BIGINT", + "INTEGER", + "BIGINT", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "DOUBLE", + "INTEGER", + "DOUBLE", + ], + 1, + ) +) +t = time.time() +con.execute( + f""" +COPY (SELECT CAST(c1/1000000 AS INTEGER) AS start_s, CAST(c2/1000000 AS INTEGER) AS end_s, + c3 AS job, c4 AS task, c5 AS machine, c6 AS cpu, c7 AS mem, c8 AS amem, c14 AS max_cpu, c19 AS agg + FROM read_csv('{G}/task_usage/part-*-of-00500.csv.gz', header=false, columns={{{cols}}}, filename=true)) +TO '{OUT}/task_usage.parquet' (FORMAT parquet, COMPRESSION zstd, ROW_GROUP_SIZE 1000000)""" +) +print("task_usage done", time.time() - t, flush=True) +jcols = "{'time':'BIGINT','missing':'INTEGER','job':'BIGINT','etype':'INTEGER','user':'VARCHAR','sclass':'INTEGER','jname':'VARCHAR','lname':'VARCHAR'}" +con.execute( + f"""COPY (SELECT job, arg_max(user, time) FILTER (WHERE user IS NOT NULL) AS user, + arg_max(lname, time) FILTER (WHERE lname IS NOT NULL) AS lname, arg_max(sclass, time) AS sclass + FROM read_csv('{G}/job_events/part-*.csv.gz', header=false, columns={jcols}) GROUP BY job) + TO '{OUT}/jobs.parquet' (FORMAT parquet)""" +) +print("jobs done", time.time() - t, flush=True) diff --git a/asap-tools/experiments/datasets/google/scripts/windows.py b/asap-tools/experiments/datasets/google/scripts/windows.py new file mode 100644 index 00000000..9afc5e2e --- /dev/null +++ b/asap-tools/experiments/datasets/google/scripts/windows.py @@ -0,0 +1,27 @@ +import sys +import duckdb + +D = sys.argv[1] +c = duckdb.connect() +c.execute(f"SET memory_limit='5GB'; SET threads=12; SET temp_directory='{D}/tmp'") +# day d covers steps [2+288d, 2+288(d+1)); 29 days exactly = 8352 steps +c.execute( + f"COPY (SELECT job, task, ((step-2)//288)::TINYINT AS dy, count(*)::SMALLINT n FROM '{D}/steps.parquet' GROUP BY ALL) TO '{D}/task_day.parquet' (FORMAT parquet)" +) +c.execute(f"CREATE TABLE td AS SELECT * FROM '{D}/task_day.parquet'") + + +def elig(d0, L, th): + need = int(th * 288 * L + 1e-9) if th < 1 else 288 * L + return c.sql( + f"""SELECT count(*) FILTER (WHERE f>=25), count(*) FILTER (WHERE f>=50), count(*) FILTER (WHERE f>=100) FROM + (SELECT job, count(*) f FROM (SELECT job, task, sum(n) s FROM td WHERE dy >= {d0} AND dy < {d0+L} GROUP BY ALL) WHERE s >= {need} GROUP BY job)""" + ).fetchone() + + +print("single days, strict: day -> jobs with >=25/50/100 full tasks") +for d in range(29): + print(d, elig(d, 1, 1.0), flush=True) +for L in (2, 3, 7, 14, 29): + for th in (1.0, 0.99): + print(f"days 0..{L} th={th}", elig(0, L, th), flush=True)