Detecting Partition Skew in H3-Sharded Streams

To detect partition skew in an H3-sharded stream, sample per-partition message counts over a fixed window, reduce that count vector to a single scalar with the coefficient of variation (stdev / mean) or a Gini coefficient, publish it as a Prometheus gauge, and alert when it stays above a threshold for several consecutive windows. A perfectly balanced topic scores zero; a single dominant partition pushes both coefficients toward their upper range.

This page belongs to Consumer Lag & Partition Skew Monitoring, part of the broader Monitoring & Observability for Spatial Pipelines reference — the section covering how to see inside a running spatial webhook pipeline before it fails.

When to use this pattern

Reach for a quantified skew metric when:

• You partition a Kafka topic by H3 cell (or any spatial key) and traffic is geographically uneven — dense urban cells generate far more events than rural ones, concentrating load on a handful of partitions.
• You need an early-warning signal that precedes consumer lag, so you can act before one consumer in a group falls permanently behind.
• You want a single numeric threshold you can alert on, rather than eyeballing a per-partition bar chart during an incident.

It is not the right tool when your topic is keyed randomly (round-robin producers spread load evenly by construction), or when you have only one or two partitions — with so few buckets the statistic is dominated by noise, and a direct per-partition lag alert is simpler and more honest.

Why H3 sharding skews, and what to measure

Sharding by H3 cell gives you locality: every event for a given hexagon lands on the same partition, so an ordered, stateful consumer sees a coherent stream per area. That locality is exactly what creates skew. Human activity is spatially clustered, so the number of events per H3 cell follows a heavy-tailed distribution — a downtown cell at resolution 7 can carry orders of magnitude more webhooks than a suburban one. When the partition assignment maps that cell to a fixed partition, the imbalance is baked into the stream. The mechanics of that mapping are covered in partitioning Kafka topics by H3 cell; this page assumes the mapping exists and focuses on measuring its fairness.

The right thing to measure is the shape of the per-partition count vector, not any single partition’s absolute rate. Two summary statistics do this well. The coefficient of variation, stdev / mean, is zero when every partition holds the same count and grows without bound as mass concentrates; it reacts sharply to a single outlier, which makes it a good alerting trigger. The Gini coefficient measures inequality on a 0–1 scale — 0 is perfect equality, 1 is total concentration in one partition — and reads more intuitively on a dashboard. Both derive from the same count vector, so compute them together.

Complete runnable implementation

The collector below samples every partition’s log-end offset twice, one window apart, differences them to get per-partition message counts, computes both skew coefficients, and exposes them as Prometheus gauges. It uses confluent-kafka for the admin/consumer offset reads and prometheus_client for the gauge. The statistics use only the standard library so the calculation is dependency-light and unit-testable in isolation.

"""
Partition-skew collector for an H3-sharded Kafka topic.

Requires: confluent-kafka>=2.0, prometheus-client>=0.19
Run as a sidecar; it scrapes counts every WINDOW_SECONDS and republishes
two gauges. Point Prometheus at the /metrics port to scrape.
"""
import time
from statistics import mean, pstdev
from typing import Sequence

from confluent_kafka import Consumer, TopicPartition
from confluent_kafka.admin import AdminClient
from prometheus_client import Gauge, start_http_server


def coefficient_of_variation(counts: Sequence[int]) -> float:
    """
    Return stdev / mean of the per-partition counts (population stdev).

    Zero for a perfectly even distribution; grows without bound as load
    concentrates on fewer partitions. Empty partitions (count 0) are kept:
    an unused partition IS the imbalance we want to surface. Returns 0.0
    when the whole topic is idle, since the mean is then undefined.
    """
    if not counts:
        return 0.0
    mu = mean(counts)
    if mu == 0:
        return 0.0
    return pstdev(counts) / mu


def gini_coefficient(counts: Sequence[int]) -> float:
    """
    Return the Gini coefficient of the per-partition counts, in [0, 1].

    0.0 = every partition carries an equal share; values near 1.0 mean a
    single partition holds almost all messages. Computed with the sorted-
    rank formula, which is O(n log n) and exact for non-negative inputs.
    """
    n = len(counts)
    total = sum(counts)
    if n == 0 or total == 0:
        return 0.0
    ordered = sorted(counts)
    # Sum of (rank-weighted) values; ranks are 1..n over the sorted vector.
    weighted = sum((i + 1) * c for i, c in enumerate(ordered))
    return (2.0 * weighted) / (n * total) - (n + 1.0) / n


def partition_counts(consumer: Consumer, topic: str,
                     partitions: Sequence[int],
                     begin: dict[int, int]) -> tuple[list[int], dict[int, int]]:
    """
    Read current log-end offsets and difference them against `begin`.

    Returns (per-partition counts for this window, new begin offsets).
    The high watermark is the offset of the next message to be produced,
    so (end - begin) is the number of messages appended during the window.
    """
    counts, new_begin = [], {}
    for p in partitions:
        tp = TopicPartition(topic, p)
        _low, high = consumer.get_watermark_offsets(tp, timeout=10.0)
        prev = begin.get(p, high)      # first pass: treat window as empty
        counts.append(max(0, high - prev))
        new_begin[p] = high
    return counts, new_begin


def list_partitions(admin: AdminClient, topic: str) -> list[int]:
    """Discover the live partition ids for the topic from cluster metadata."""
    md = admin.list_topics(topic, timeout=10.0)
    if topic not in md.topics or md.topics[topic].error is not None:
        raise RuntimeError(f"topic {topic!r} not found in cluster metadata")
    return sorted(md.topics[topic].partitions.keys())


def run(bootstrap: str, topic: str, window_seconds: int = 300,
        metrics_port: int = 9109) -> None:
    admin = AdminClient({"bootstrap.servers": bootstrap})
    consumer = Consumer({
        "bootstrap.servers": bootstrap,
        "group.id": "skew-collector",      # never commits; read-only
        "enable.auto.commit": False,
    })

    cov_gauge = Gauge(
        "h3_partition_skew_cov",
        "Coefficient of variation of per-partition message counts",
        ["topic"],
    )
    gini_gauge = Gauge(
        "h3_partition_skew_gini",
        "Gini coefficient of per-partition message counts",
        ["topic"],
    )

    start_http_server(metrics_port)
    begin: dict[int, int] = {}
    while True:
        parts = list_partitions(admin, topic)
        counts, begin = partition_counts(consumer, topic, parts, begin)
        cov_gauge.labels(topic=topic).set(coefficient_of_variation(counts))
        gini_gauge.labels(topic=topic).set(gini_coefficient(counts))
        time.sleep(window_seconds)


if __name__ == "__main__":
    # Adjust bootstrap servers and topic to your cluster.
    run(bootstrap="localhost:9092", topic="spatial-events-h3")

The first loop iteration seeds the begin offsets and reports zero skew (its window is empty by definition); every subsequent iteration reports the true per-window distribution.

Parameter reference

Gotchas and spatial edge cases

1. Transient versus structural skew. A one-off spatial burst — a stadium emptying, a storm cell crossing a dense metro — spikes the coefficient for a few windows, then subsides. Structural skew persists because the underlying H3 cell distribution is permanently uneven. Only structural skew justifies re-sharding. Gate alerts on k consecutive windows above the threshold so a single burst never triggers action.

2. Empty partitions must stay in the vector. It is tempting to drop zero-count partitions before computing statistics, but an idle partition is precisely the imbalance you are hunting. Filtering zeros makes a badly skewed topic look balanced. Keep every partition; only special-case a total count of zero (an idle topic), where the mean is undefined and both coefficients return 0.0.

3. Choosing the window is a latency trade-off. Too short and normal per-key bursts read as skew, generating false pages; too long and a genuinely hot partition accumulates consumer lag before the metric moves. Five to fifteen minutes suits most webhook streams. If you also run a short lag alert, you can afford a longer skew window because lag catches the fast-moving failures.

4. Offset gaps from compaction or retention. On a log-compacted topic, or one whose retention deletes segments mid-window, end - begin overstates the count because deleted offsets still advance the high watermark. For skew detection this rarely matters — the relative shape is preserved — but do not treat the counts as an exact message tally. For an accurate produced-message count, prefer a compaction-free topic or read broker-side per-partition produce metrics.

5. Re-sharding is expensive, so confirm before you act. Changing the H3 resolution or the partition count remaps every key and forces a full consumer-group rebalance, replaying state for stateful consumers. The skew coefficient is a leading indicator; confirm sustained consumer lag on the hot partition before paying that cost. A high coefficient with healthy lag means your consumers are absorbing the imbalance fine.

6. CoV is unbounded, Gini is not — do not compare them. The two coefficients answer different questions. A CoV of 2.4 and a Gini of 0.7 can describe the same vector. Alert on one (CoV, for its sharp outlier response) and chart the other (Gini, for its 0–1 readability); never set a single threshold that mixes them.

Verification

Unit-test the pure statistics against distributions with known answers — a flat vector must score zero, and a fully concentrated vector must approach the maximum. Run with pytest:

import math
import pytest
from your_module import coefficient_of_variation, gini_coefficient


def test_even_distribution_has_zero_skew():
    """A perfectly balanced topic scores zero on both coefficients."""
    counts = [100, 100, 100, 100]
    assert coefficient_of_variation(counts) == 0.0
    assert gini_coefficient(counts) == 0.0


def test_single_hot_partition_is_maximally_skewed():
    """One partition holding all traffic pushes both metrics to their peak."""
    counts = [0, 0, 0, 400]
    # CoV of [0,0,0,x] is sqrt(n-1) = sqrt(3) for n = 4.
    assert coefficient_of_variation(counts) == pytest.approx(math.sqrt(3))
    # Gini approaches (n-1)/n = 0.75 for total concentration in one bucket.
    assert gini_coefficient(counts) == pytest.approx(0.75)


def test_empty_partitions_are_counted_not_filtered():
    """Zero-count partitions raise skew rather than being ignored."""
    balanced = gini_coefficient([50, 50, 50, 50])
    with_idle = gini_coefficient([50, 50, 50, 50, 0, 0])
    assert with_idle > balanced


def test_idle_topic_is_defined_and_zero():
    """A topic with no traffic must not divide by zero."""
    assert coefficient_of_variation([0, 0, 0]) == 0.0
    assert gini_coefficient([0, 0, 0]) == 0.0


def test_known_uneven_distribution():
    """A hand-checked vector matches the closed-form Gini value."""
    counts = [1, 2, 3, 4]           # total 10, n = 4
    # weighted = 1*1 + 2*2 + 3*3 + 4*4 = 30
    # gini = 2*30 / (4*10) - 5/4 = 1.5 - 1.25 = 0.25
    assert gini_coefficient(counts) == pytest.approx(0.25)

The test_single_hot_partition_is_maximally_skewed case is the important one: it pins both coefficients to their closed-form extremes on a known-uneven distribution, so a future refactor cannot silently break the calculation.

FAQ

Related

• Consumer Lag & Partition Skew Monitoring — the parent guide pairing this skew metric with consumer-lag tracking
• Partitioning Kafka Topics by H3 Cell — the sharding scheme whose fairness this page measures
• H3 vs S2 vs Quadkey for Spatial Partitioning — how the grid choice shapes the skew you will observe