实验 13：在线聚合（Online Aggregation）

为什么做这个实验

`SELECT max(b) FROM stuff` 这样的聚合查询，传统做法是扫完全表才吐出一个数。表小的时候无所谓，
一旦数据到亿行级、或者扫描本身要走磁盘、走网络，用户就只能对着一个没有任何反馈的界面干等。

可是多数分析场景里，用户并不是要一个"精确到最后一位"的数字，他只是想知道**大概是多少**，
好决定是继续等、还是换个查询条件重来。在线聚合就是为此设计的：扫描每推进一段，就把当前估计值
**连同它可能偏离真值多少**一起报出来，用户可以随时喊停。

本实验要弄清楚三件事：

1. 把"一次性返回结果"的聚合改造成"逐行产出结果"，只需要把 `return` 换成 `yield`（生成器）。
2. 只有**均值型**聚合才配得上置信区间；`MAX` 的运行值是一个确定的下界而不是估计值，两者不能混为一谈。
3. 置信区间公式成立有一个硬性前提：**扫描顺序必须是随机的**。这一条被破坏时，误差条会整体错位，
而且加宽区间也救不回来。

原始论文：Hellerstein、Haas、Wang，《Online Aggregation》，SIGMOD 1997。该文提出在线聚合三要素：
运行估计（running estimate）、置信区间（confidence bound）、停止规则（stopping rule）。
本 notebook 用生成器实现第 1 项；第 2、3 项的公式与实测数字见末尾小节。

运行前准备

• Python 3.10 及以上。

• 依赖：`faker`（造演示数据）、`numpy` 与 `matplotlib`（复现末尾的置信区间与 1/√m 收缩，选做）、`jupyter`。

• 缺包时安装：

```bash
python -m pip install faker numpy matplotlib
```

（本机已配置环境可用：`python -m pip install faker numpy matplotlib`）

• 打开方式：在仓库根目录执行 `jupyter notebook notebooks/Online-Aggregation.ipynb`。

• 规模提示：本 notebook 只造 **40 行** faker 数据，用于理解接口，**不产生统计意义**；

末尾引用的统计结论来自 n = 200 000 的独立实测。

要解决的问题

```sql
SELECT max(b) FROM stuff
```

这条查询有两种实现方式，函数体几乎一行不差，区别只在**什么时候把结果交出去**：

• **全量聚合（full-blown aggregation）**：扫到最后一行才 `return` 一个值。在此之前用户什么都看不到。

• **在线聚合（online aggregation）**：每扫过一行就 `yield` 一次当前值。用户处理完第 1 行就拿到第一个结果，

之后每个结果都比上一个更接近真值。

"更接近真值"这句话对不同的聚合函数含义完全不同，后面会专门拆开讲：

| 聚合 | 运行值的性质 | 能否配置信区间 |
| --- | --- | --- |
| `MAX` / `MIN` | 单调变化的**确定下界**（MAX 只升不降） | 不给分布假设就写不出 |
| `AVG` / `SUM` / `COUNT` | 基于已扫描样本的**估计值** | 可以，见末尾公式 |

下面先用 40 行小数据把两种写法跑通。

第 1 步：造一批演示数据

`Faker` 用来生成像人名的假数据；`fake.seed_instance(42)` 固定随机种子，保证**每次运行看到的数据一样**，
便于对照与排错。`Stuff` 是一个只有 `a`、`b` 两个整数字段的数据类，用来模拟关系里的一行。

看输出：`data[:10]` 会打印 10 个 `Stuff(a=..., b=...)`。你只需要确认两件事——
数据确实造出来了，并且换台机器再跑一次结果相同。

from faker import Faker

fake = Faker()
fake.seed_instance(42)

from dataclasses import dataclass

number_of_tuples = 40


@dataclass
class Stuff:
    a: int
    b: int


# create some fake data mimicking tuples in a relation:
data = [
    Stuff(fake.pyint(max_value=1000), fake.pyint(max_value=2000))
    for f in range(number_of_tuples)
]
data[:10]

[Stuff(a=654, b=228),
 Stuff(a=25, b=1518),
 Stuff(a=281, b=501),
 Stuff(a=228, b=285),
 Stuff(a=754, b=209),
 Stuff(a=692, b=1516),
 Stuff(a=913, b=1116),
 Stuff(a=89, b=1209),
 Stuff(a=432, b=65),
 Stuff(a=30, b=191)]

第 2 步：算法 1 —— 全量聚合

要做什么：从头到尾扫一遍，维护"目前见到的最大值"，**扫完才返回**。

为什么这样写：

• `current_max = -math.inf`：初值取负无穷，保证任何真实数据都能把它顶掉。若初值写 0，

而数据全是负数，结果就会错成 0。

• 状态只有一个标量，所以空间复杂度是 **O(1)**，与数据量无关。

预期看到：一个单独的整数，就是这 40 行里 `b` 的最大值。
它也是后续在线聚合最后一个 `yield` 的值——单元测试正是断言这两者相等。

import math


def aggregate(input_data, attribute):
    """Full-blown aggregation algorithm: computes an aggregation over a list of tuples, once it is done, it returns the result"""
    current_max = -math.inf
    for el in input_data:
        current_max = max(current_max, el.__dict__[attribute])
    return current_max


aggregate(data, "b")

第 3 步：算法 2 —— 在线聚合

要做什么：把 `return` 换成循环里的 `yield`，函数就从"一次性交结果"变成"逐行交结果"。

为什么这样写：

• 带 `yield` 的函数被调用时**不执行函数体**，而是返回一个生成器（generator）；

每次 `for` 取一个值，函数才往前走到下一个 `yield`。这正是"边扫边报"的最小骨架。

• 函数体与算法 1 逐行相同，说明在线化的代价不在计算量，而在**接口形态**。

预期看到：40 行输出，数值只升不降（这是 `MAX` 的性质），最后一个数与算法 1 的输出完全一致。

两个初学者常踩的坑：

1. **生成器只能消费一次**，取完了再遍历是空的。要重看得重新调用 `online_aggregation(...)`，
或者先 `list(...)` 缓存下来。
2. 生成器是**惰性**的：不迭代它，函数体一行都不会跑。所以"改成 yield 就自动变快"是误解——
它变的是"什么时候能拿到结果"，不是"算得快不快"。

import math


def online_aggregation(input_data, attribute):
    """Online aggregation algorithm: computes a running aggregation over a list of tuples yielding the new result after each tuple"""
    current_max = -math.inf
    for el in input_data:
        current_max = max(current_max, el.__dict__[attribute])
        yield current_max


for agg in online_aggregation(data, "b"):
    print(agg)

第 4 步：单元测试——两条路径必须殊途同归

单元测试断言：对随机造出的 1000 行数据，全量聚合的结果与在线聚合的**最后一个**产出值相等。

关键写法是 `*_, res2 = online_aggregation(self.data, "b")`：

• `*_` 把前面所有值丢进一个占位变量（下划线表示"我不用"）；

• `res2` 接到最后一个 `yield` 的值。

如果输入为空，生成器一次也不 `yield`，这行会直接报解包错误——这也是"空输入要单独处理"的一个提醒。

跑完应看到 `Ran 1 test ... OK`。测试通过说明：改成流式接口**没有改变语义**，两种写法给出的答案一致。

import unittest


class MyTest(unittest.TestCase):

    # the following function will be called before every test-function:

    def create_data(self, number_of_tuples):
        self.data = [
            Stuff(fake.pyint(max_value=20), fake.pyint(max_value=20))
            for f in range(number_of_tuples)
        ]

    def test_scans_end_to_end(self):
        for _ in range(10):
            self.create_data(1000)

            res1 = aggregate(self.data, "b")
            *_, res2 = online_aggregation(self.data, "b")

            self.assertEqual(res1, res2)


# Run the unit test without shutting down the jupyter kernel
# notice that this call will look for all test classes, i.e.
# all classes inheriting from unittest.TestCase
# each of these functions will be executed once
# unittest.main(argv=['ignored', '-v'], verbosity=2, exit=False)

# only execute a specific Test-class:
unittest.main(argv=["ignored", "-v"], defaultTest="MyTest", verbosity=2, exit=False)

test_scans_end_to_end (__main__.MyTest.test_scans_end_to_end) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.021s

OK

<unittest.main.TestProgram at 0x1103bd2b0>

停下来想清楚：`MAX` 与 `AVG` 的待遇为什么不同

**`MAX` 的运行值是确定的下界。** 已经扫过的最大值不会再变小，真值只可能更大。所以看到"当前 1937"，
用户可以断言"真值至少 1937"，这个断言不需要任何概率论。但他**不能**断言"真值就是 1937"——
除非对 `b` 的分布另有假设，否则最后 1% 里藏着一个巨值的可能性无法排除。
因此 `MAX` 写不出统计意义上的置信区间，它提供的是"至少有多大"，不是"有多大把握"。

**`AVG` 不一样。** 已扫描的 m 行可以看成总体的一个样本，样本均值围绕真值波动，波动幅度可以量化：

| 符号 | 含义 |
| --- | --- |
| m | 已扫描行数 |
| x̄ₘ | 样本均值，(x₁ + x₂ + … + xₘ) / m，即对真值的当前估计 |
| s | 这 m 行的样本标准差（总体标准差未知，用它代替） |
| s² | 样本方差 |
| SE | 标准误差（standard error），SE = √(s²/m) = s/√m |
| 1.96 | 标准正态分布的 95% 分位数 |

95% 置信区间（confidence interval，CI）：

[ x̄ₘ − 1.96 · SE , x̄ₘ + 1.96 · SE ]

区间宽度：

W(m) = 2 × 1.96 × s / √m

要点是 **W 与 √m 成反比**：想让误差减半，m 要扩大到 4 倍；从扫 1% 到扫 100%，宽度只缩到 1/10。
也就是说，精度越往后面越"贵"。

本机实测（n = 200 000，Gamma 分布数据，随机顺序扫描，估计对象为均值型聚合，95% CI）：

| 扫描进度 m/n | 运行估计 | 说明 |
| --- | --- | --- |
| 10% | 50.1674 | 已与真值相差不到 0.2 |
| 100% | 49.9817 | 即总体真值 |
| 100% | — | 95% CI 宽度 0.3090 |

（口径：本机实测，CPython 3.13 + Windows；数字以本案例库《00-实测数字口径表》为准。
你在本机跑出的具体值随随机种子略有差异，判断标准是**数量级与趋势**，不是逐位一致。）

由宽度公式反推总体标准差：s ≈ W × √m / 3.92 = 0.3090 × √200000 / 3.92 ≈ 35.2，
与所用 Gamma 分布（shape = 2、scale = 25，理论均值 50、标准差约 35.4）吻合。

三要素（运行估计、置信区间、停止规则）的提法出自 Hellerstein、Haas、Wang，《Online Aggregation》，SIGMOD 1997。

一个不能跳过的前提：扫描顺序必须是随机的

置信区间公式把"已扫描的 m 行"当作总体的**随机样本**。这个前提一旦破坏，公式就失效，
而且失效方式比"估得不准"严重得多：

• 表恰好按 `b` **升序**存放时，扫前 10% 拿到的是最小的 10% 的值，估计系统性偏低；

• **降序**则系统性偏高；

• 更糟的是这个偏差**不随 m 增大而消失**：扫到 99% 时，样本仍然缺着最大的那一小撮值。

这时的误差叫**偏差（bias）**，不是方差。置信区间是围绕"无偏估计 ± 1.96·SE"构造的，
中心本身就错了，无论把区间加多宽都罩不住真值。

工程含义：在线聚合要么要求存储层提供**随机顺序扫描**（random-order scan），要么在读取时打乱顺序。
物理布局不受你控制时，这一点必须打个问号。

（提醒：本 notebook 的 40 行 faker 数据规模太小，看不出上述效应。要观察它，请在 n = 200 000 量级上
分别用随机置换顺序与 `np.sort(values)` 升序顺序生成两趟扫描，比较 m = 10% 时的估计值与 CI 是否罩住真值。）

别认错：哪些"长得像"的技术不是在线聚合

判断标准只有一条：**有没有同时给出"估计值"和"误差条"**。

• **DuckDB 的查询进度条**：输出已完成的百分比和预计剩余时间（ETA）。它回答的是"还要等多久"，

不是"答案大概是多少"。它不改变估计的统计性质，也不做抽样推断——在线聚合三要素里它一项都不占，
**不是**在线聚合。

• **BigQuery 的 `APPROX_*` 函数**（如 `APPROX_COUNT_DISTINCT`）：基于 sketch（概要数据结构）的

一次遍历近似，用精度换速度。它通常不随时间流式细化，也一般不附带置信区间，属于**另一条技术路线**，
文献引用时不要与在线聚合混为一谈。

一句话区分两者交换的东西：

• `APPROX_*` 换的是**精度**：一次遍历、常数内存给出有误差保证的近似值，运行的时间结构没变；

• 在线聚合换的是**时间结构**：扫描可以随时中断，中断点越靠后答案越准，单行成本未必更省。

换句话说：`APPROX_*` 是"算得快但不能边算边看"，在线聚合是"能边算边看但未必算得快"。

思考题与详解

**1. `aggregate` 与 `online_aggregation` 的函数体几乎一模一样，只差一个 `yield`。请从内存占用与首结果延迟（time to first result）两个角度比较两者。**

内存：两者都是 O(1)——都只维护一个"当前最大值"标量，与数据量无关，所以差别**不在内存**。

首结果延迟：`aggregate` 必须扫完 n 行才有第一个（也是唯一一个）结果；`online_aggregation` 处理完
第 1 行就产出第一个结果。n = 2×10⁵ 时前者要等整趟扫描，后者的第一个结果几乎立刻就到。

更重要的是接口形态：流式接口允许调用方在任意时点**放弃**剩余扫描（"止损"），这正是停止规则
（stopping rule）能实现的接口基础。注意别把"首结果快"理解成"总耗时短"——生成器是惰性的，
总计算量一行也没少。

</details>

**2. `MAX` 的运行值单调不减、且永远不大于真值。为什么它不需要置信区间也能用？用户看到"运行值 1937 且 5 分钟没变"能推出什么、不能推出什么？**

运行值是**确定性的下界**：无论剩下的数据是什么，真值只可能 ≥ 1937，所以"至少 1937"这个命题永远成立，
不需要任何概率论与分布假设——这也是它不需要置信区间的原因。

5 分钟没变，只能推出"这 5 分钟里没遇到更大的值"，**不能**推出"真值就是 1937"。
除非对 `b` 的分布另有假设（例如已知上界、或已知尾部极薄），否则最后 1% 里藏着一个巨值的可能性
无法排除。

对比 `AVG`：均值型估计必须配 CI，因为它既可能高也可能低，没有"单向保证"可用。

</details>

**3. 由实测的 W(100%) = 0.3090（n = 200 000）反推数据的样本标准差 s；若要把 CI 宽度压到 0.1，需要扫多少行？在这张表上做得到吗？**

反推：W = 2 × 1.96 × s / √m ⇒ s = W × √m / 3.92 = 0.3090 × √200000 / 3.92 ≈ **35.2**
（与 Gamma(2, 25) 的理论标准差 √(2×25²) ≈ 35.4 吻合）。

压到 0.1：m = (3.92 × s / 0.1)² = (3.92 × 35.2 / 0.1)² ≈ **1.9 × 10⁶ 行**，约为本表行数（2×10⁵）的 9.5 倍。
**在这张表上做不到**，只能换更大的表或接受更宽的区间。

这说明 1/√m 的收缩非常"贵"：**精度要求每提高 10 倍，样本量要扩大 100 倍**。在线聚合的价值
在"前几十个百分点里把误差快速压下来"，而不是追求任意高的精度。

</details>

**4. 为什么升序扫描让置信区间**失效**，而不只是"估计偏低"？提示：CI 公式的哪一条前提被破坏了？**

CI 公式的前提是"已扫描的 m 行构成总体的一个**随机样本**"，于是样本均值是**无偏**估计，
误差只表现为方差，幅度由 SE 刻画。

升序扫描时，前 m 行是全体中最小的 m 个值，样本均值系统性偏低，且这个偏低**不随 m 增大而消失**
（扫到 99% 仍然缺着最大的 1%）。这是**偏差**，不是方差。

置信区间是按"无偏估计 ± 1.96·SE"构造的：中心本身就错了，区间整体错位，无论加多宽都罩不住真值。
偏差无法靠加宽区间弥补——这正是"随机顺序扫描不是可选项"的原因。

</details>

**5. 把置信水平从 95% 提到 99%，分位数由 1.96 换成约 2.58。同样扫 20 万行，CI 宽度变为多少倍？精度与把握之间怎么取舍？**

宽度与分位数成正比：2.58 / 1.96 ≈ **1.32 倍**。也就是把握提高 4 个百分点，代价是区间放宽约三成
（由 0.3090 变成约 0.41）。

取舍取决于用途：探索性报表、初筛可以用 90%（分位数约 1.64），窄区间更好读；
涉及钱、承诺、对外披露的数字要 99% 以上，或者干脆扫完拿精确值。

注意：提高把握和"多扫数据"是两种独立手段。想同时变窄又更有把握，只能增大 m。

</details>

**6. DuckDB 的查询进度条与在线聚合的本质区别是什么？如果一个系统只输出"已完成 37%"，用户能对最终结果做出什么判断、不能做出什么判断？**

本质区别：进度条提供的是**执行状态**（扫到哪了、还要多久），在线聚合提供的是**结果的统计性质**
（答案大概是多少、误差多大）。前者不涉及抽样与推断，后者以"估计值 + 误差条"为核心。

看到"已完成 37%"，用户**能**判断"还要等多久"（配合 ETA）；
**不能**判断"当前已经看到的部分结果离最终答案差多少"——因为系统压根没给出部分结果。

在线聚合三要素（运行估计、置信区间、停止规则，Hellerstein、Haas、Wang，SIGMOD 1997）
进度条一项都不具备，所以它**不是**在线聚合。

</details>

**7. BigQuery 的 `APPROX_COUNT_DISTINCT` 与在线聚合都在"没扫完就要答案"，两者各自交换出去的是什么？**

`APPROX_*` 交换的是**精度**：基于 sketch（概要结构）一次遍历、常数内存给出带误差保证的近似值。
它的时间结构与精确计算相同（都是扫一遍），只是每一步更省，**不随扫描进度细化**，
也通常不附置信区间。

在线聚合交换的是**时间结构**：扫描可以随时中断、答案随中断点单调变准（CI 越来越窄），
单行处理成本未必更省。

一句话：`APPROX_*` 是"算得快但不能边算边看"，在线聚合是"能边算边看但未必算得快"。
两者是不同技术路线，写文献综述时不要互相混引。

</details>

**8. 停止规则实验里，目标宽度取 0.3 时，停止位置反而比理论推算（约 106% 的表）更晚甚至直接扫完。解释为什么目标宽度接近 W(n) 时提前停止失去意义。**

W(n) = 0.3090 是**扫完全表时**的宽度下限（m 最大，区间最窄）。任何中途时刻 m < n，
都有 W(m) = W(n)·√(n/m) > W(n)。

因此若目标宽度 ≤ W(n)（如 0.3 < 0.3090），停止条件 `W(m) ≤ 目标` 在扫完之前**永远不成立**，
停止规则退化成全表扫描，"提前停止"失去意义。
按公式反解目标 0.3 需要 m ≈ (0.3090/0.3)² × n ≈ 1.06 n，超过全表行数，所以必然扫完。

工程含义：用停止规则之前先算一遍理论可达精度 W(n)，或者让系统在"扫完仍不达标"时给出明确提示，
而不是静默地跑完整趟扫描。

</details>