Building Score Aggregation Pipelines #

Problem Framing #

A per-URL health score answers "is this page okay?" It cannot answer "is the checkout section degrading?" or "did last week's release move the site composite?" Those questions require rolling thousands of individual scores up into a small number of trustworthy composites — one per section, one per template, one for the whole site. Get the rollup wrong and the failure is invisible: a section score looks stable because five well-performing URLs are diluting forty broken ones, or a weekly recompute silently drops a template because a schema change renamed a column upstream. Teams that skip a deliberate aggregation layer end up debugging dashboards instead of pages.

This guide builds that aggregation layer end to end: the grouping and coverage logic, the configuration surface that makes rollups reproducible, the daily-plus-weekly execution schedule, and the artifact contract downstream consumers rely on. It assumes per-URL composite scores already exist — for how those scores are computed in the first place, see Designing Custom Health Score Algorithms. This work sits inside the broader Metric Scoring & Data Normalization topic area, which covers normalization, thresholding, and trend analysis alongside aggregation.

---

Prerequisites & Environment Setup #

The rollup job reads scored Parquet output, so it depends on the same object storage and Python runtime as the per-URL scoring pipeline, plus a coverage reference — the full list of known URLs per section — so it can tell "no page here scored below 0.5" apart from "no page here was scored at all."

Pinned tool versions

Required environment variables

# /etc/aggregation-pipeline/env  — sourced by the cron wrapper
export SCORED_URL_BUCKET="gs://site-audit-artifacts-prod/scored"
export ROLLUP_OUTPUT_BUCKET="gs://site-audit-artifacts-prod/rollups"
export URL_INVENTORY_TABLE="postgresql://scorer:${SCORING_DB_PASS}@db.internal:5432/audit_metrics.url_inventory"
export ROLLUP_CONFIG="/opt/aggregation/rollup_config.json"
export TZ="UTC"

Confirm the URL inventory table is current before the first run — coverage calculations are only as accurate as the denominator. A section with 40 known URLs where 38 were scored has 95% coverage; the same 38 scored URLs against a stale inventory of 20 known URLs would wrongly report 100%.

---

Step 1 — Initialization: The Aggregation Job #

Load the most recent scored records for each URL, join them against the URL inventory to know which section and template each belongs to, and group by the configured rollup key. Do this grouping before any weighting logic runs — weighting operates within a group, never across groups.

# /opt/aggregation/rollup.py
# Requires: pandas==2.2.2, pyarrow==16.0.0, numpy==1.26.4
from __future__ import annotations
import json
import os
from pathlib import Path

import numpy as np
import pandas as pd
import pyarrow.parquet as pq


def load_latest_scores(scored_bucket: str, window_days: int = 1) -> pd.DataFrame:
    """Load per-URL composite scores written by the scoring pipeline."""
    cutoff = pd.Timestamp.utcnow() - pd.Timedelta(days=window_days)
    frames = []
    for path in Path(scored_bucket).glob("**/*.parquet"):
        table = pq.read_table(path)
        df = table.to_pandas()
        df = df[pd.to_datetime(df["scored_at"], utc=True) >= cutoff]
        if not df.empty:
            frames.append(df)
    if not frames:
        raise ValueError(f"No scored records found in the last {window_days} day(s)")
    return pd.concat(frames, ignore_index=True)


def join_url_inventory(scores: pd.DataFrame, inventory: pd.DataFrame) -> pd.DataFrame:
    """
    Attach section, template, and monthly traffic share to every scored URL.
    A left join preserves scored URLs that inventory metadata briefly lags behind.
    """
    merged = scores.merge(inventory, on="url", how="left", validate="one_to_one")
    unmatched = merged["section"].isna().sum()
    if unmatched:
        print(f"WARNING: {unmatched} scored URLs missing inventory metadata")
    return merged


def group_for_rollup(merged: pd.DataFrame, rollup_level: str) -> pd.core.groupby.DataFrameGroupBy:
    """Group scored, inventory-joined rows by the configured rollup key."""
    key_by_level = {
        "section": "section",
        "template": "template",
        "site": None,  # single group across the whole DataFrame
    }
    key = key_by_level[rollup_level]
    if key is None:
        merged = merged.assign(_site_key="site")
        return merged.groupby("_site_key")
    return merged.groupby(key)

Reject rows with a null url or a composite score outside [0.0, 1.0] before grouping — a single corrupted score inside a small section can move that section's composite more than a handful of individually flawed pages would in a larger group.

---

Step 2 — Core Configuration: Rollup Parameters #

Three parameters decide most of the rollup's behavior: which key to group by, how much coverage a group needs before its composite is trusted, and whether URLs are weighted equally or by their share of traffic. Keep them in a versioned config file rather than as script arguments, so a rollup can always be reproduced from its config hash alone.

Key parameters table

// /opt/aggregation/rollup_config.json  (version-controlled)
{
  "version": "1.4.0",
  "rollup_level": "section",
  "coverage": {
    "coverage_floor": 0.70,
    "min_sample_size": 5,
    "stale_after_days": 3
  },
  "weighting": {
    "weight_by_traffic": true,
    "aggregation_method": "weighted_mean",
    "traffic_window_days": 30
  },
  "schedule": {
    "incremental": "daily",
    "full_recompute": "weekly"
  }
}

# /opt/aggregation/aggregate.py
# Requires: pandas==2.2.2, numpy==1.26.4
from __future__ import annotations

import numpy as np
import pandas as pd


def compute_group_composite(
    group: pd.DataFrame,
    total_known_urls: int,
    config: dict,
) -> dict:
    """Aggregate one rollup group (a section, template, or the whole site)."""
    coverage_cfg = config["coverage"]
    weight_cfg = config["weighting"]

    fresh = group[group["age_days"] <= coverage_cfg["stale_after_days"]]
    coverage_ratio = len(fresh) / total_known_urls if total_known_urls else 0.0

    if coverage_ratio < coverage_cfg["coverage_floor"] or len(fresh) < coverage_cfg["min_sample_size"]:
        return {"composite_score": None, "coverage_ratio": coverage_ratio, "status": "low_confidence"}

    if weight_cfg["weight_by_traffic"]:
        weights = fresh["traffic_share"].clip(lower=1e-6)
    else:
        weights = pd.Series(1.0, index=fresh.index)
    weights = weights / weights.sum()

    if weight_cfg["aggregation_method"] == "weighted_geometric_mean":
        composite = float(np.exp((np.log(fresh["composite_score"].clip(lower=1e-9)) * weights).sum()))
    else:
        composite = float((fresh["composite_score"] * weights).sum())

    return {"composite_score": composite, "coverage_ratio": coverage_ratio, "status": "ok"}

The status field is as important as the score itself. Downstream alerting and dashboards must treat low_confidence groups differently — surfacing the coverage gap rather than a misleadingly precise number. See Percentile Normalization Across Metric Distributions for how rolled-up composites are further ranked against a reference distribution once they clear the coverage floor.

---

Rollup Pipeline Architecture #

The diagram below traces a scored URL from the per-URL scoring output through inventory join, coverage evaluation, weighted aggregation, and into the partitioned rollup artifact that downstream dashboards and alerting read.

---

Step 3 — Execution & Scheduling #

Run the rollup twice on separate cadences. The daily incremental run aggregates only URLs scored since the last run, keeping section composites current within a day. The weekly full recompute reprocesses the entire scoring window from scratch, which is the only way to pick up a rollup_config.json change or backfill a previously low_confidence group once its coverage recovers. Both share the same flock lock as the scoring pipeline's artifact bucket to avoid writing conflicting partitions.

#!/usr/bin/env bash
# /opt/aggregation/run_rollup.sh
set -euo pipefail

MODE="${1:-incremental}"   # incremental | full
LOCKFILE="/var/lock/site-health-rollup.lock"
LOG="/var/log/site-health-rollup/$(date -u +%Y%m%d-%H%M%S).log"
SCRIPT_DIR="/opt/aggregation"

exec 200>"${LOCKFILE}"
flock -n 200 || { echo "Rollup already running — exiting." >&2; exit 1; }

mkdir -p "$(dirname "${LOG}")"

{
  echo "=== Rollup run (${MODE}) started at $(date -u +%Y-%m-%dT%H:%M:%SZ) ==="

  export CONFIG_HASH
  CONFIG_HASH=$(sha256sum "${ROLLUP_CONFIG}" | awk '{print $1}')
  echo "CONFIG_HASH=${CONFIG_HASH}"

  if [ "${MODE}" = "full" ]; then
    WINDOW_DAYS=90
  else
    WINDOW_DAYS=1
  fi

  python3 "${SCRIPT_DIR}/rollup.py" \
    --config "${ROLLUP_CONFIG}" \
    --window-days "${WINDOW_DAYS}" \
    --config-hash "${CONFIG_HASH}" \
    --mode "${MODE}"

  echo "=== Rollup run (${MODE}) completed at $(date -u +%Y-%m-%dT%H:%M:%SZ) ==="
} 2>&1 | tee "${LOG}"

Cron entries — /etc/cron.d/site-health-rollup:

# Daily incremental rollup at 04:30 UTC, after per-URL scoring at 03:15
30 4 * * * scoring-user bash /opt/aggregation/run_rollup.sh incremental

# Full weekly recompute Sunday at 05:00 UTC, after the weekly full score run
0 5 * * 0   scoring-user bash /opt/aggregation/run_rollup.sh full

Pin TZ=UTC for both the cron daemon's environment and the stale_after_days age calculation inside rollup.py — a rollup running under a local timezone can misjudge which scores are "fresh" near midnight, silently shifting which URLs count toward coverage. If the job runs inside CI instead of cron, use a workflow-level concurrency key against the same rollup output prefix so parallel triggers do not race the same partition, following the same pattern used when integrating custom crawlers with CI/CD pipelines for artifact-writing jobs.

---

Step 4 — Artifact Capture & Storage #

Write rollup output as date-partitioned Parquet, one file per rollup level per run, tagged with the config hash and run mode so any historical composite can be traced back to the exact configuration that produced it.

# /opt/aggregation/artifact.py
# Requires: pyarrow==16.0.0, pandas==2.2.2
from __future__ import annotations
from datetime import datetime, timezone

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq


def write_rollup_artifact(
    rollup_df: pd.DataFrame,
    rollup_level: str,
    config_hash: str,
    mode: str,
    bucket: str,
    run_date: datetime | None = None,
) -> str:
    run_date = run_date or datetime.now(tz=timezone.utc)
    partition = run_date.strftime("%Y/%m/%d")

    metadata = {
        b"rollup_level": rollup_level.encode(),
        b"config_hash": config_hash.encode(),
        b"mode": mode.encode(),
        b"run_timestamp": run_date.isoformat().encode(),
    }

    table = pa.Table.from_pandas(rollup_df).replace_schema_metadata(metadata)
    path = f"{bucket}/{rollup_level}/{partition}/rollup_{mode}_{config_hash[:8]}.parquet"
    pq.write_table(table, path, compression="snappy")
    print(f"Rollup artifact written: {path}")
    return path

Partition by rollup level first, then date — rollups/section/2026/07/05/... — so a dashboard querying only section-level history never has to scan template- or site-level partitions. Retain daily incremental artifacts for 90 days and full-recompute artifacts for 400 days, since the weekly full runs are what auditors and year-over-year trend reports actually query. The lifecycle rules that automate this expiry — rather than a hand-rolled purge script — are covered in Storing & Versioning Crawl Artifacts in Cloud Storage.

For a rollup that also needs to reflect uneven measurement across URL path prefixes rather than the section taxonomy alone, see Aggregating Scores Across URL Segments, which extends this same coverage-and-weighting approach to arbitrary path-based groupings.

---

Verification Checklist #

Run these checks after every rollup before treating its output as authoritative for dashboards or alerting:

1. Confirm the log contains === Rollup run (<mode>) completed at — an early exit (lock contention or an unhandled exception) leaves stale data as the latest artifact.
2. Verify row counts: the number of groups in the output equals the number of distinct sections (or templates) in the URL inventory, plus any that were marked low_confidence rather than silently dropped: python3 -c "import pandas as pd; df=pd.read_parquet('path/to/rollup.parquet'); print(df['status'].value_counts())".
3. Assert every non-null composite_score is bounded to [0.0, 1.0] — a value outside that range means an upstream per-URL score escaped its own validation.
4. Check the config_hash embedded in the artifact's Parquet metadata matches sha256sum "${ROLLUP_CONFIG}" at run time, confirming no local edit slipped past version control.
5. Diff the current run's composite scores against the prior run for the same rollup level: a section moving by more than 0.10 without a corresponding deployment event or traffic-mix shift merits investigation before it feeds an alert.
6. Spot-check one low_confidence group by hand — pull its raw scored URLs and confirm the coverage ratio the job computed matches a manual count against the URL inventory.

---

Troubleshooting #

Section composites disappear entirely after a schema change

Root cause: the URL inventory table renamed or restructured the section column, so the inventory join in join_url_inventory silently produces all-null section values, and every group collapses into a single NaN bucket.

# Confirm section coverage in the joined DataFrame before grouping
python3 -c "
import pandas as pd
df = pd.read_parquet('/tmp/joined_scores.parquet')
print(df['section'].isna().mean())
"

Fix: update the join key in rollup.py to match the current inventory schema, and add a hard assertion that fails the run if more than 1% of scored URLs have a null section rather than letting them silently merge into an unlabeled group.

---

Every section reports low_confidence after a normal run

Root cause: stale_after_days is set tighter than the actual scoring cadence, so URLs scored two days ago (well within a normal daily cycle) are excluded from the "fresh" count before the coverage ratio is even computed.

# Compare the configured staleness window against the real scoring cadence
python3 -c "
import json
cfg = json.load(open('/opt/aggregation/rollup_config.json'))
print('stale_after_days:', cfg['coverage']['stale_after_days'])
"

Fix: set stale_after_days to at least twice the scoring job's actual interval (3 days for a daily scorer gives one full missed-run buffer), then re-run the incremental rollup.

---

Traffic-weighted composite diverges sharply from the unweighted average

Root cause: traffic_share values were computed over a stale or mis-scoped window — often because the traffic source still reflects last month's data after a migration, so one or two pages dominate the weighting.

# Sum traffic_share per section — it should total close to 1.0 within tolerance
python3 -c "
import pandas as pd
df = pd.read_parquet('/tmp/joined_scores.parquet')
print(df.groupby('section')['traffic_share'].sum())
"

Fix: confirm the traffic source's traffic_window_days matches rollup_config.json, refresh the traffic extract, and re-run the full recompute rather than patching the incremental run in place.

---

Weekly full recompute times out before completing

Root cause: the full recompute reprocesses the entire 90-day window in a single in-memory pandas pass, and the scored-URL volume has grown past what fits comfortably without spilling.

# Check the row count the full recompute is attempting to load
python3 -c "
import pyarrow.parquet as pq
print(pq.read_metadata('gs://site-audit-artifacts-prod/scored/').num_rows)
"

Fix: chunk the full recompute by rollup group (process one section's history at a time) instead of loading the entire scored dataset into a single DataFrame, or move the aggregation step to a query engine that pushes the group-by down before materializing results.

---

Rollup artifact for a section is present but never appears on the dashboard

Root cause: the dashboard reads only the latest partition per rollup level, and a delayed or retried run wrote its output under yesterday's date partition instead of today's, because run_date was not explicitly pinned to the run's logical date.

# List today's and yesterday's section partitions
gsutil ls "gs://site-audit-artifacts-prod/rollups/section/$(date -u +%Y/%m/%d)/"
gsutil ls "gs://site-audit-artifacts-prod/rollups/section/$(date -u -d yesterday +%Y/%m/%d)/"

Fix: pass an explicit run_date matching the scheduling window into write_rollup_artifact rather than defaulting to wall-clock time, so retried or delayed runs still land in the correct logical partition.

---

Config hash in the artifact does not match the config file in version control

Root cause: a local, uncommitted edit to rollup_config.json was used for a production run, usually while testing a new coverage_floor value.

# Compare the committed config hash against the one embedded in the last artifact
git show HEAD:rollup_config.json | sha256sum
python3 -c "
import pyarrow.parquet as pq
m = pq.read_metadata('path/to/rollup.parquet').metadata
print(m[b'config_hash'])
"

Fix: commit the validated config, re-run the full recompute using the committed version, and treat any config change as a reviewable pull request rather than a direct edit on the production host.

---

Common Mistakes #

• Grouping by section before joining against a current URL inventory, which locks in whatever section labels existed at the last inventory refresh rather than the site's current structure.
• Treating a low_confidence group as a score of zero on a dashboard instead of surfacing it distinctly — this creates false alarms that erode trust in the whole rollup layer.
• Running the daily incremental job without a shared lock against the weekly full recompute, producing a partition with mixed partial and complete data.
• Weighting by traffic without a floor on the minimum weight per URL, letting a single near-zero-traffic page's traffic_share collapse to a value so small it effectively disappears from a small section's composite.
• Skipping the rollup-level partitioning scheme and writing every level into one flat prefix, forcing every downstream query to filter instead of prune partitions.

---

What happens to a section when too few URLs were scored to aggregate confidently?

The rollup checks the fraction of known URLs in that section that received a fresh score in the current window against coverage_floor. If coverage falls below the floor, the section is marked low_confidence instead of being assigned a composite score, which prevents a handful of stale or missing pages from silently producing a misleading section-level number.

Why weight by traffic instead of taking a simple average across URLs in a section?

An unweighted average lets a rarely visited utility page move a section's score as much as a top landing page. Traffic weighting means the composite reflects the experience most real visitors actually have, which is the number that should drive alerting and prioritization decisions.

Can the daily incremental rollup and the weekly full recompute conflict with each other?

They write to the same artifact prefix, so both runs acquire the same flock lock before starting. If the weekly full recompute is still running when the daily incremental job would normally fire, the incremental run exits immediately rather than racing the full recompute, and the next scheduled incremental run picks up any records it missed.

What rollup levels are supported besides section?

The same job supports template-level rollups (grouping by the page template a URL renders with, such as product or category) and a single site-level composite. Set rollup_level to section, template, or site in the config; each level reuses the identical coverage and weighting logic, only the group-by key changes.

---

Related #

• Metric Scoring & Data Normalization — parent topic area covering the full normalization and scoring workflow
• Aggregating Scores Across URL Segments — coverage-weighted aggregation by path-based segment instead of section taxonomy
• Designing Custom Health Score Algorithms — how the per-URL composite scores fed into this rollup are computed
• Percentile Normalization Across Metric Distributions — ranking rolled-up composites against a rolling reference distribution