Pipeline requirements: SLAs, freshness, sizing

"""Thirty days of Campus Bite's daily payout file, checked against its two targets.

Each night a job starts at 04:30, reads the payment provider's settled orders of the day before
and publishes the payout file for the canteens. The targets are measured where the file is read:

    fresh:    the file is published by 06:00
    complete: it holds at least 99.9 % of the day's settled orders, counted from the provider's
              own settlement report, not from what the job happened to read

The runs are synthetic and seeded, so every run of this program prints the same 30 days.
"""

import random

SEED = 20261017
DAYS = 30
START = 4 * 60 + 30  # the job starts at 04:30 (minutes after midnight)
DEADLINE = 6 * 60  # published by 06:00
COMPLETE = 99.9  # per cent of the settled orders
TARGET_DAYS = 29  # each target must hold on at least 29 of the 30 days


def clock(minutes):
    return f"{minutes // 60:02d}:{minutes % 60:02d}"


def simulate(rng):
    days = []
    for day in range(1, DAYS + 1):
        settled = 38_000 + int(rng.random() * 6_000)
        # Most nights every settled order has arrived before 04:30; on some, part of the
        # provider's file comes later and the job never sees it.
        late = int(settled * (0.002 + rng.random() * 0.006)) if rng.random() < 0.08 else 0
        # Most runs take 28 to 52 minutes; a slow night takes 85 to 110.
        minutes = 85 + int(rng.random() * 26) if rng.random() < 0.02 else 28 + int(rng.random() * 25)
        days.append({"day": day, "settled": settled, "in_file": settled - late, "published": START + minutes})
    return days


def main():
    days = simulate(random.Random(SEED))
    print(f"Campus Bite payout file: {DAYS} nightly runs (seeded)")
    print(f"Targets: published by {clock(DEADLINE)}, holding at least {COMPLETE} % of the day's settled orders")
    print()
    print("Days that missed a target:")
    on_time = complete = both = 0
    shares = []
    for d in days:
        share = 100 * d["in_file"] / d["settled"]
        shares.append(share)
        fresh = d["published"] <= DEADLINE
        full = share >= COMPLETE
        on_time += fresh
        complete += full
        both += fresh and full
        if not (fresh and full):
            missing = d["settled"] - d["in_file"]
            what = f"{missing:,} of {d['settled']:,} orders missing" if missing else "nothing missing"
            print(f"  day {d['day']:>2}: published {clock(d['published'])}, {share:6.2f} % complete ({what})")
    print()
    print(f"The job finished without an error:      {DAYS} of {DAYS} days")
    print(f"Published by {clock(DEADLINE)}:                     {on_time} of {DAYS} days, target {TARGET_DAYS}: {'met' if on_time >= TARGET_DAYS else 'missed'}")
    print(f"At least {COMPLETE} % complete:              {complete} of {DAYS} days, target {TARGET_DAYS}: {'met' if complete >= TARGET_DAYS else 'missed'}")
    print(f"Fresh and complete, as a reader saw it: {both} of {DAYS} days")
    print(f"Completeness averaged over the month:   {sum(shares) / DAYS:.2f} %")


main()

Campus Bite payout file: 30 nightly runs (seeded)
Targets: published by 06:00, holding at least 99.9 % of the day's settled orders

Days that missed a target:
  day  6: published 05:11,  99.21 % complete (332 of 42,255 orders missing)
  day 18: published 05:10,  99.69 % complete (126 of 41,304 orders missing)
  day 21: published 05:21,  99.27 % complete (288 of 39,270 orders missing)
  day 24: published 06:07, 100.00 % complete (nothing missing)
  day 26: published 05:16,  99.45 % complete (235 of 42,590 orders missing)
  day 28: published 05:11,  99.72 % complete (114 of 40,406 orders missing)

The job finished without an error:      30 of 30 days
Published by 06:00:                     29 of 30 days, target 29: met
At least 99.9 % complete:              25 of 30 days, target 29: missed
Fresh and complete, as a reader saw it: 24 of 30 days
Completeness averaged over the month:   99.91 %

"""Bytes per order in three layouts: CSV, JSON Lines and a column-wise packed binary.

The program makes 200,000 synthetic orders of the invented Campus Bite platform (seeded, so
every run prints the same numbers), writes them in three layouts into memory and counts the
bytes. No general-purpose compressor is used: the packed layout only stores each column on
its own, numbers as binary, repeated names once, and times as the seconds since the previous
order.
"""

import csv
import io
import json
import random
from bisect import bisect
from itertools import accumulate

SEED = 20261017
ORDERS = 200_000
DAYS = 5
FIRST_DAY = "2031-03-04"  # the synthetic data's first day (UTC); the next four follow it
CAMPUSES = ["north", "south", "east", "west", "river", "hill", "lake", "park", "city", "valley", "garden", "harbour"]
STATUSES = ["delivered", "cancelled", "refunded"]
STATUS_SHARE = list(accumulate([92, 5, 3]))
# Share of the day's orders placed in each hour (breakfast, a lunch rush, dinner).
HOUR_SHARE = list(accumulate([1, 1, 0, 0, 0, 1, 2, 4, 7, 6, 5, 8, 14, 13, 6, 4, 4, 5, 7, 9, 8, 5, 3, 2]))
COLUMNS = ["order_id", "placed_at", "campus", "canteen", "rider", "status", "items", "amount"]


def arrival(r):
    """When one order is placed, in seconds from the start of the first day."""
    day = int(r() * DAYS)
    hour = bisect(HOUR_SHARE, r() * HOUR_SHARE[-1])
    # Lectures end at 12:00 and 13:00: half of those two hours' orders come in their first 10 minutes.
    if hour in (12, 13) and r() < 0.5:
        second = int(r() * 600)
    else:
        second = int(r() * 3_600)
    return day * 86_400 + hour * 3_600 + second


def make_orders(rng):
    """200,000 orders in time order. placed_at is in seconds from the start of the first day;
    amounts are whole numbers of the smallest currency unit."""
    r = rng.random
    times = sorted(arrival(r) for _ in range(ORDERS))
    orders = []
    for n, t in enumerate(times):
        status = STATUSES[bisect(STATUS_SHARE, r() * STATUS_SHARE[-1])]
        orders.append({
            "order_id": 1_000_001 + n,
            "placed_at": t,
            "campus": CAMPUSES[int(r() * len(CAMPUSES))],
            "canteen": 1 + int(r() * 60),
            "rider": 0 if status == "cancelled" else 1 + int(r() * 2_400),
            "status": status,
            "items": 1 + int(r() * 6),
            "amount": (30 + int(r() * 571)) * 100,
        })
    return orders


def stamp(t):
    """An ISO 8601 UTC time for t seconds after the start of the first day."""
    day, rest = divmod(t, 86_400)
    year, month, first = FIRST_DAY.split("-")
    return f"{year}-{month}-{int(first) + day:02d}T{rest // 3600:02d}:{rest // 60 % 60:02d}:{rest % 60:02d}Z"


def as_csv(orders, stamps):
    out = io.StringIO()
    w = csv.writer(out, lineterminator="\n")
    w.writerow(COLUMNS)
    w.writerows([o["order_id"], s, o["campus"], o["canteen"], o["rider"], o["status"], o["items"], o["amount"]] for o, s in zip(orders, stamps))
    return out.getvalue().encode("utf-8")


def as_json_lines(orders, stamps):
    encode = json.JSONEncoder(separators=(",", ":")).encode
    lines = [encode(dict(o, placed_at=s)) for o, s in zip(orders, stamps)]
    return ("\n".join(lines) + "\n").encode("utf-8")


def put_varint(out, n):
    """An unsigned number in 7-bit groups: 0-127 take one byte, 128-16,383 two, and so on."""
    while n >= 0x80:
        out.append((n & 0x7F) | 0x80)
        n >>= 7
    out.append(n)


def deltas(values):
    """The first value, then each value minus the one before it, every one as a varint."""
    out = bytearray()
    put_varint(out, values[0])
    for before, now in zip(values, values[1:]):
        put_varint(out, now - before)
    return bytes(out)


def dictionary(values):
    """The distinct names once, then one byte per row that says which name it is."""
    names = sorted(set(values))
    index = {name: i for i, name in enumerate(names)}
    head = json.dumps(names).encode("utf-8")
    out = bytearray()
    put_varint(out, len(head))
    return bytes(out) + head + bytes(index[v] for v in values)


def varints(values):
    out = bytearray()
    for v in values:
        put_varint(out, v)
    return bytes(out)


def as_packed(orders):
    """Column by column, each column a block of bytes with its length in front."""
    col = {name: [o[name] for o in orders] for name in COLUMNS}
    blocks = {
        "order_id": deltas(col["order_id"]),
        "placed_at": deltas(col["placed_at"]),
        "campus": dictionary(col["campus"]),
        "canteen": bytes(col["canteen"]),
        "rider": b"".join(r.to_bytes(2, "little") for r in col["rider"]),
        "status": dictionary(col["status"]),
        "items": bytes(col["items"]),
        "amount": varints(col["amount"]),
    }
    out = bytearray()
    head = json.dumps({"rows": len(orders), "columns": COLUMNS}).encode("utf-8")
    put_varint(out, len(head))
    out += head
    for block in blocks.values():
        put_varint(out, len(block))
        out += block
    return bytes(out), blocks


def main():
    orders = make_orders(random.Random(SEED))
    stamps = [stamp(o["placed_at"]) for o in orders]
    csv_bytes = as_csv(orders, stamps)
    json_bytes = as_json_lines(orders, stamps)
    packed, blocks = as_packed(orders)

    print(f"{ORDERS:,} synthetic Campus Bite orders over {DAYS} days (seeded)")
    print()
    print("The first order in the two text layouts:")
    print("  CSV         " + csv_bytes.split(b"\n")[1].decode())
    print("  JSON Lines  " + json_bytes.split(b"\n")[0].decode())
    print()
    print(f"{'layout':<18}{'bytes':>12}{'per order':>11}{'vs CSV':>8}")
    for name, data in [("CSV", csv_bytes), ("JSON Lines", json_bytes), ("packed by column", packed)]:
        print(f"{name:<18}{len(data):>12,}{len(data) / ORDERS:>11.1f}{len(data) / len(csv_bytes):>8.2f}")
    print()
    print("Bytes per order in the packed layout, column by column:")
    for name, block in blocks.items():
        print(f"  {name:<10}{len(block) / ORDERS:>5.2f}")
    print()
    print(f"CSV takes {len(csv_bytes) / len(packed):.1f} times the bytes of the packed layout,")
    print(f"and JSON Lines {len(json_bytes) / len(packed):.1f} times.")


main()

200,000 synthetic Campus Bite orders over 5 days (seeded)

The first order in the two text layouts:
  CSV         1000001,2031-03-04T00:00:07Z,harbour,17,0,cancelled,4,17900
  JSON Lines  {"order_id":1000001,"placed_at":"2031-03-04T00:00:07Z","campus":"harbour","canteen":17,"rider":0,"status":"cancelled","items":4,"amount":17900}

layout                   bytes  per order  vs CSV
CSV                 11,989,127       59.9    1.00
JSON Lines          28,789,067      143.9    2.40
packed by column     2,153,376       10.8    0.18

Bytes per order in the packed layout, column by column:
  order_id   1.00
  placed_at  1.00
  campus     1.00
  canteen    1.00
  rider      2.00
  status     1.00
  items      1.00
  amount     2.77

CSV takes 5.6 times the bytes of the packed layout,
and JSON Lines 13.4 times.

"""When Campus Bite's orders arrive: the hour-by-hour profile, the busiest minute and the peak factor.

The arrival times are made exactly as in layout_sizes.py (the same seed, the same draws), so these
are the same 200,000 synthetic orders. The peak factor is the busiest minute's orders divided by
the orders of an average minute: the number a sizing estimate multiplies the average by.
"""

import random
from bisect import bisect
from collections import Counter
from itertools import accumulate

SEED = 20261017
ORDERS = 200_000
DAYS = 5
EVENTS_PER_ORDER = 6  # placed, paid, accepted, ready, picked up, delivered
HOUR_SHARE = list(accumulate([1, 1, 0, 0, 0, 1, 2, 4, 7, 6, 5, 8, 14, 13, 6, 4, 4, 5, 7, 9, 8, 5, 3, 2]))
PER_HASH = 5  # orders a minute that one # stands for


def arrival(r):
    """When one order is placed, in seconds from the start of the first day."""
    day = int(r() * DAYS)
    hour = bisect(HOUR_SHARE, r() * HOUR_SHARE[-1])
    # Lectures end at 12:00 and 13:00: half of those two hours' orders come in their first 10 minutes.
    if hour in (12, 13) and r() < 0.5:
        second = int(r() * 600)
    else:
        second = int(r() * 3_600)
    return day * 86_400 + hour * 3_600 + second


def main():
    r = random.Random(SEED).random
    times = sorted(arrival(r) for _ in range(ORDERS))
    minutes = DAYS * 24 * 60
    average = ORDERS / minutes  # orders in an average minute

    by_hour = Counter(t // 3_600 % 24 for t in times)
    by_minute = Counter(t // 60 for t in times)
    busiest, most = max(sorted(by_minute.items()), key=lambda kv: kv[1])
    day, rest = divmod(busiest, 24 * 60)

    print(f"When {ORDERS:,} synthetic orders are placed ({DAYS} days, seeded)")
    print()
    print(f"Orders a minute, hour by hour, averaged over the {DAYS} days (each # is {PER_HASH} orders a minute):")
    for hour in range(24):
        rate = by_hour[hour] / (DAYS * 60)
        print(f"  {hour:02d}:00 {rate:6.1f}  {'#' * round(rate / PER_HASH)}")
    top_hour = max(range(24), key=lambda h: by_hour[h])
    top_rate = by_hour[top_hour] / (DAYS * 60)
    factor = most / average
    print()
    print(f"Average minute:          {average:5.1f} orders")
    print(f"Busiest hour, a minute:  {top_rate:5.1f} orders ({top_hour:02d}:00 to {top_hour + 1:02d}:00), {top_rate / average:.1f} times the average")
    print(f"Busiest minute:          {most:5d} orders (day {day + 1} at {rest // 60:02d}:{rest % 60:02d}), {factor:.1f} times the average")
    print()
    print(f"Peak factor for sizing: {factor:.1f}")
    print(f"With {EVENTS_PER_ORDER} events an order: {average * EVENTS_PER_ORDER / 60:.1f} events a second on average, "
          f"at most {most * EVENTS_PER_ORDER / 60:.1f} in the busiest minute")


main()

When 200,000 synthetic orders are placed (5 days, seeded)

Orders a minute, hour by hour, averaged over the 5 days (each # is 5 orders a minute):
  00:00    5.8  #
  01:00    6.0  #
  02:00    0.0  
  03:00    0.0  
  04:00    0.0  
  05:00    5.6  #
  06:00   11.2  ##
  07:00   22.8  #####
  08:00   40.4  ########
  09:00   35.5  #######
  10:00   28.4  ######
  11:00   46.5  #########
  12:00   80.7  ################
  13:00   76.1  ###############
  14:00   34.7  #######
  15:00   23.4  #####
  16:00   23.3  #####
  17:00   28.6  ######
  18:00   40.9  ########
  19:00   52.3  ##########
  20:00   46.0  #########
  21:00   28.9  ######
  22:00   17.6  ####
  23:00   11.9  ##

Average minute:           27.8 orders
Busiest hour, a minute:   80.7 orders (12:00 to 13:00), 2.9 times the average
Busiest minute:            311 orders (day 4 at 12:03), 11.2 times the average

Peak factor for sizing: 11.2
With 6 events an order: 2.8 events a second on average, at most 31.1 in the busiest minute

"""Storage and peak load for Campus Bite's order events, from inputs you can change.

Change the numbers in SCENARIOS and run it again. Every figure it prints is a multiplication
or a division of those inputs:

    events a day        = orders a day x events per order
    raw bytes a day     = events a day x bytes per event
    stored bytes a day  = raw bytes a day / compression ratio
    stored at the end   = stored bytes a day x retention days (older days are deleted)
    average per second  = events a day / 86,400
    peak per second     = average per second x peak factor
"""

SECONDS_A_DAY = 86_400
MB = 1_000_000  # decimal units, as storage is sold: 1 GB is 1,000,000,000 bytes
GB = 1_000_000_000

# Where the inputs come from:
#   40,000 orders a day: the synthetic term days of layout_sizes.py and peaks.py (200,000 orders in 5 days).
#   6 events an order (placed, paid, accepted, ready, picked up, delivered), each carrying the whole order
#   row, so an event is the size of one row as JSON Lines: 144 bytes in layout_sizes.py.
#   13.4: how many times smaller layout_sizes.py measured the same rows packed by column.
#   400 days: a year of payouts for disputes, plus a month to spare.
#   11.2: the busiest minute against the average minute, measured by peaks.py. Using it for all six
#   events of an order is on the safe side: an order's later events come minutes after it is placed.
SCENARIOS = [
    # name,                        orders a day, events per order, bytes per event, compression, days, peak factor
    ("This year, as JSON Lines",         40_000,                6,             144,         1.0,  400,        11.2),
    ("This year, packed by column",      40_000,                6,             144,        13.4,  400,        11.2),
    ("Next year: twice the orders",      80_000,                6,             144,        13.4,  400,        11.2),
]


def size(value):
    """Bytes as MB below a gigabyte and as GB above it."""
    return f"{value / GB:,.2f} GB" if value >= GB else f"{value / MB:,.1f} MB"


for name, orders, per_order, nbytes, ratio, days, peak_factor in SCENARIOS:
    events = orders * per_order
    raw_a_day = events * nbytes
    stored_a_day = raw_a_day / ratio
    average = events / SECONDS_A_DAY
    print(name)
    print(f"  {orders:,} orders x {per_order} events x {nbytes} bytes, compression {ratio}x, kept {days} days")
    print(f"  events a day             {events:>10,}")
    print(f"  raw data a day           {size(raw_a_day):>10}")
    print(f"  stored a day             {size(stored_a_day):>10}")
    print(f"  stored after 30 days     {size(stored_a_day * 30):>10}")
    print(f"  stored after {days} days    {size(stored_a_day * days):>10}")
    print(f"  average per second       {average:>7.1f} events")
    print(f"  peak per second ({peak_factor}x)  {average * peak_factor:>7.1f} events")
    print()

print("Storage grows until the oldest day is deleted; the consumers must keep up with the peak.")

This year, as JSON Lines
  40,000 orders x 6 events x 144 bytes, compression 1.0x, kept 400 days
  events a day                240,000
  raw data a day              34.6 MB
  stored a day                34.6 MB
  stored after 30 days        1.04 GB
  stored after 400 days      13.82 GB
  average per second           2.8 events
  peak per second (11.2x)     31.1 events

This year, packed by column
  40,000 orders x 6 events x 144 bytes, compression 13.4x, kept 400 days
  events a day                240,000
  raw data a day              34.6 MB
  stored a day                 2.6 MB
  stored after 30 days        77.4 MB
  stored after 400 days       1.03 GB
  average per second           2.8 events
  peak per second (11.2x)     31.1 events

Next year: twice the orders
  80,000 orders x 6 events x 144 bytes, compression 13.4x, kept 400 days
  events a day                480,000
  raw data a day              69.1 MB
  stored a day                 5.2 MB
  stored after 30 days       154.7 MB
  stored after 400 days       2.06 GB
  average per second           5.6 events
  peak per second (11.2x)     62.2 events

Storage grows until the oldest day is deleted; the consumers must keep up with the peak.