# 实验 15：分布式连接

> 两张表放在不同的机器上时，用什么办法把它们连起来，以及"要搬多少数据"该怎么算。

## 一、实验目标

1. 区分两种数据分布方式：**垂直分区**（每个节点存整张表）与**水平分区**（每个节点存每张表的一部分）。
2. 掌握三种数据移动策略：整表传输、广播连接（broadcast join）、半连接归约（semi-join reduction）。
3. 理解协同分区（co-partition）：按连接键（join key）分区时，连接可以在各节点本地完成。
4. 会用"传输字节数"这个度量比较不同策略，并按选择率、节点数、主键宽度做敏感性分析。
5. 验证一件事：不同执行策略得到的**结果集必须完全相同**，差异只在代价。

## 二、预备知识

### 2.1 分布式数据库的基本设定

**分布式 DBMS**（distributed DBMS）把数据放在多个节点（node / server）上，节点之间通过网络通信。与单机 DBMS 相比，多了一项关键代价：**网络传输**。局域网带宽通常比磁盘顺序读慢一到两个数量级，因此分布式查询优化的第一原则是"少搬数据"。

### 2.2 垂直分区与水平分区

**垂直分区**（vertical partitioning）：一张表整个放在一个节点上，不同表在不同节点。

```
   节点 s1                    节点 s2
┌──────────────┐         ┌──────────────┐
│  persons     │         │  books       │
│  (整张表)     │         │  (整张表)     │
└──────────────┘         └──────────────┘
        └──────── 网络 ────────┘
```

**水平分区**（horizontal partitioning）：每张表按某个键切成若干份，分散到各节点。

```
   节点 s1                          节点 s2
┌────────────────────┐         ┌────────────────────┐
│ persons (A–L)      │         │ persons (M–Z)      │
│ books   (A–L)      │         │ books   (M–Z)      │
└────────────────────┘         └────────────────────┘
        按连接键 author/name 的姓氏首字母切分 → 协同分区
```

**协同分区**（co-partitioning）：两张表按**同一个连接键**、用**同一个哈希/范围函数**分区。这样"能连上的记录一定在同一个节点"，连接可以完全在本地完成。

### 2.3 半连接（semi-join）

**半连接**（semi-join）：`R ⋉ S` 的结果只保留 R 中那些"在 S 里能找到匹配"的记录，且不重复、不带 S 的列。

半连接归约的思路：先把一边的**连接键投影**（很小）发过去，用它把另一边的无用记录过滤掉，再传输过滤后的数据。因为它只发"键"，所以比发整张表省得多。

```
   s1 (有 persons)                              s2 (有 books)
   1. SELECT DISTINCT name
      WHERE birthday < 2001-01-01
   2. ────── 只发连接键投影（很小） ──────────▶
                                                3. books ⋉ 这些键（半连接，本地过滤）
   5. ◀────── 只发过滤后的 books ────────────── 4.
   6. 本地做最终连接
```

### 2.4 三种数据移动策略

| 策略 | 做法 | 适合场景 |
|---|---|---|
| 整表传输 | 把一张表整个搬到另一张表所在节点 | 表小、或者对方节点算力冗余 |
| 广播连接 | 把小表复制到**每一个**有大事实的节点 | 小表足够小，且要保留并行度 |
| 半连接归约 | 先发连接键投影，再发过滤后的数据 | 过滤条件选择率低、连接键投影远小于整行 |

真实系统的相关默认值（可用于对照）：Spark 的 `spark.sql.autoBroadcastJoinThreshold` 默认 10485760（10 MB），AQE（自适应查询执行）自 Spark 3.2.0 起默认开启；Trino 的 `join-distribution-type` 默认 AUTOMATIC（PARTITIONED 两侧重分布、BROADCAST 要求右表过滤后能放进每个节点内存、无统计信息时回退到哈希分布式）。

## 三、实验原理

### 3.1 实验用的查询

```sql
SELECT *
FROM persons JOIN books
     ON name = author
WHERE birthday < 2001-01-01
```

- `persons`：5,000 条，每条约 60 字节（本模型取值）
- `books`：10,000 条，每条约 52 字节（本模型取值）
- 节点数：3
- 过滤条件 `birthday < 2001-01-01` 的选择率假设为 60%（即 60% 的 persons 满足）

注意：`persons` 与 `books` 的连接键是**姓名**（`name = author`），不是整数主键。这一点对传输量影响很大（见思考题 3）。

### 3.2 三种策略的传输量

**① 整表传输**（把 books 整表从 s2 发到 s1）

```
传输量 = |books| × 每条字节数 = 10,000 × 52 B = 520,000 B ≈ 508 KB
```

**② 广播连接**（把 books 复制到 3 个节点）

```
传输量 = 508 KB × 3 ≈ 1,523 KB
```

**③ 半连接归约**

```
第 1 步：发 persons 的连接键投影（16 B/条，未按 birthday 过滤的全量 5,000 条）  = 5,000 × 16 B      =  80,000 B
第 2 步：发过滤后的 books（10,000 × 60% × 52 B）                 = 312,000 B
合计                                                            = 392,000 B ≈ 383 KB
```

三种策略的对比图：

![分布式连接的三种数据移动策略](../figures/fig_distjoin.png)

按本模型计算，半连接归约比整表传输少约 25%，比广播连接少约 75%。

**与仓库 notebook 的对照**：`notebooks/Distributed-Joins.ipynb` 里用 `table_size()` 逐行按字符串实际长度统计，报告的数字是"整表约 522 KB，半连接约 71 + 316 = 387 KB"。与上面的模型值（508 KB / 383 KB）量级一致，差异来自"每条记录按固定字节估算"还是"按字符串实际长度逐行统计"。两种口径都可以用，但**比较时必须统一口径**。

### 3.3 水平分区下的本地连接

如果两表已按连接键协同分区，那么每个节点可以独立完成自己那份连接，最后把两个局部结果合并即可：

```
s1: persons(A–L) ⋈ books(A–L)  ─┐
                                 ├─▶ 合并（UNION）→ 最终结果
s2: persons(M–Z) ⋈ books(M–Z)  ─┘
```

此时网络传输的只有**最终（或部分）结果**，通常是三种做法里最省的一种。代价从"搬输入"变成了"搬输出"。

## 四、实验环境与准备

- Python ≥ 3.9；必做部分只用标准库。
- 选做部分（复现 notebook）需要 `duckdb`、`pandas`、`faker`：

```bash
python -m pip install duckdb pandas faker
```

- 对应 notebook：`notebooks/Distributed-Joins.ipynb`。
- 若无法安装 duckdb/faker，只做步骤 1、2、5 即可完成本实验的核心内容。

## 五、实验步骤

### 步骤 1：建立传输量计算模型（必做）

把三种策略写成函数，方便后面做敏感性分析。

```python
KB = 1024

def size_kb(n_rows, bytes_per_row):
    """按"条数 × 每条字节数"估算表的大小，返回 KB。"""
    return n_rows * bytes_per_row / KB


def strategy_whole_table(n_books, book_bytes):
    """① 整表传输：把 books 整表发到 persons 所在节点。"""
    return size_kb(n_books, book_bytes)


def strategy_broadcast(n_books, book_bytes, n_workers):
    """② 广播连接：把 books 复制到每个节点。"""
    return size_kb(n_books, book_bytes) * n_workers


def strategy_semi_join(n_persons, n_books, book_bytes, key_bytes, selectivity):
    """③ 半连接归约：先发连接键投影，再发过滤后的 books。"""
    keys = n_persons * key_bytes                       # 第 1 步
    reduced_books = n_books * selectivity * book_bytes  # 第 2 步
    return (keys + reduced_books) / KB


# 本实验的规模
N_PERSONS, N_BOOKS = 5_000, 10_000
PERSON_BYTES, BOOK_BYTES = 60, 52
N_WORKERS = 3
KEY_BYTES = 16
SELECTIVITY = 0.6

a = strategy_whole_table(N_BOOKS, BOOK_BYTES)
b = strategy_broadcast(N_BOOKS, BOOK_BYTES, N_WORKERS)
c = strategy_semi_join(N_PERSONS, N_BOOKS, BOOK_BYTES, KEY_BYTES, SELECTIVITY)
print(f"① 整表传输   : {a:8.1f} KB")
print(f"② 广播连接   : {b:8.1f} KB")
print(f"③ 半连接归约 : {c:8.1f} KB")
print(f"半连接 / 整表 = {c / a:.2f}    半连接 / 广播 = {c / b:.2f}")
```

本机输出：

```
① 整表传输   :    507.8 KB
② 广播连接   :   1523.4 KB
③ 半连接归约 :    382.8 KB
半连接 / 整表 = 0.75    半连接 / 广播 = 0.25
```

### 步骤 2：敏感性分析（必做）

改变选择率、节点数、连接键宽度，看结论会不会翻转。

```python
print("选择率对半连接的影响（整表固定为 %.0f KB）：" % a)
for sel in [0.1, 0.2, 0.4, 0.6, 0.8, 1.0]:
    v = strategy_semi_join(N_PERSONS, N_BOOKS, BOOK_BYTES, KEY_BYTES, sel)
    print(f"  选择率 {sel:4.1f} → {v:7.1f} KB {'（比整表省）' if v < a else '（不如整表）'}")

print("\n节点数对广播连接的影响：")
for w in [2, 3, 5, 8, 16]:
    print(f"  节点 {w:2d} → {strategy_broadcast(N_BOOKS, BOOK_BYTES, w):7.1f} KB")

print("\n连接键宽度对半连接的影响（选择率 0.6）：")
for k in [4, 8, 16, 32]:
    print(f"  键宽 {k:2d} B → {strategy_semi_join(N_PERSONS, N_BOOKS, BOOK_BYTES, k, SELECTIVITY):7.1f} KB")
```

把结果抄进观测记录表，并回答一个问题：**选择率到多少时，半连接归约就不再比整表传输划算？**（用 `a == c` 解一下临界值。）

### 步骤 3：用 duckdb 复现 notebook 的两种做法（选做）

下面这段代码与 `notebooks/Distributed-Joins.ipynb` 等价，用内存 duckdb 模拟三个节点。

```python
import duckdb
import pandas as pd
from faker import Faker

Faker.seed(42)
fake = Faker()
number_of_tuples = 10000        # notebook 口径：两表同为 10,000 条
# 注意：3.1 节的传输量模型是 persons = 5,000 / books = 10,000，与这里的两表
# 10,000 条不是同一口径，notebook 实测不能与模型值逐位对齐

persons = [(fake.name(), str(fake.date())) for _ in range(number_of_tuples)]
books = [(fake.sentence(), persons[i % number_of_tuples][0],
          fake.pyfloat(min_value=0, max_value=100, right_digits=2))
         for i in range(number_of_tuples)]
persons_df = pd.DataFrame(persons, columns=["name", "birthday"])
books_df = pd.DataFrame(books, columns=["title", "author", "price"])


def table_size(table: pd.DataFrame) -> float:
    """估算 DataFrame 的大小（KB）：字符串按长度计，其余按 4 字节计。"""
    size = 0
    for _, row in table.iterrows():
        for e in row:
            size += len(e) if isinstance(e, str) else 4
    return size / 1024


class Server:
    """一个节点：内含一个 duckdb 连接和一块本地数据区。"""
    def __init__(self):
        self.con = duckdb.connect(database=":memory:")
        self.local_data = {}

    def import_data(self, tables):
        for table_name, df in tables.items():
            self.con.register(table_name, df)
            self.con.execute(f"CREATE TABLE {table_name} AS SELECT * FROM {table_name}")

    def execute_query(self, query, parameters, name):
        df = self.con.execute(query, parameters).fetchdf()
        self.local_data[name] = df
        self.con.register(name, df)

    def get_local_data(self, name):
        return self.local_data.get(name)

    def receive(self, name, df):
        self.con.register(name, df)

    def send(self, other, local_data_name):
        other.receive(local_data_name, self.get_local_data(local_data_name))
```

**做法 1.1：整表传输**

```python
s1, s2 = Server(), Server()
s1.import_data({"persons": persons_df})
s2.import_data({"books": books_df})

s2.execute_query("SELECT * FROM books", [], "books_local")
s2.send(s1, "books_local")
print("发送 books 整表，大小 %.0f KB" % table_size(s2.get_local_data("books_local")))

s1.execute_query("""SELECT * FROM persons JOIN books_local ON name = author
                    WHERE birthday < ?""", ["2001-01-01"], "result")
print("结果行数:", s1.get_local_data("result").shape[0])
```

**做法 1.2：半连接归约**

```python
s1.execute_query("SELECT DISTINCT name FROM persons WHERE birthday < ?",
                 ["2001-01-01"], "persons_filtered")
s1.send(s2, "persons_filtered")
print("发送连接键投影，大小 %.0f KB" % table_size(s1.get_local_data("persons_filtered")))

s2.execute_query("""SELECT books.* FROM books JOIN persons_filtered
                    ON author = name""", [], "books_filtered")
s2.send(s1, "books_filtered")
print("发送过滤后的 books，大小 %.0f KB" % table_size(s2.get_local_data("books_filtered")))

s1.execute_query("""SELECT * FROM persons JOIN books_filtered ON name = author
                    WHERE persons.birthday < ?""", ["2001-01-01"], "result")
print("结果行数:", s1.get_local_data("result").shape[0])
```

### 步骤 4：结果一致性校验

三种做法（整表传输、半连接归约、本地连接）必须得到**完全相同**的结果集，否则说明某个环节写错了。

```python
def normalized(df):
    return df.sort_values(by=df.columns.tolist()).reset_index(drop=True)

r1 = normalized(result_entire_table)
r2 = normalized(result_semi_join)
pd.testing.assert_frame_equal(r1, r2)
print("整表传输 与 半连接归约 的结果一致：", r1.shape)
```

### 步骤 5：水平分区 + 本地连接（选做）

按姓氏首字母把两表协同分区（< 'M' 走 s1，≥ 'M' 走 s2），各自本地连接后合并。

```python
def partition(df, column):
    mask = df[column].str.split(" ").str[1].str[0] < "M"
    return df[mask].reset_index(drop=True), df[~mask].reset_index(drop=True)

p1_persons, p2_persons = partition(persons_df, "name")
p1_books, p2_books = partition(books_df, "author")

s1, s2 = Server(), Server()
s1.import_data({"persons": p1_persons, "books": p1_books})
s2.import_data({"persons": p2_persons, "books": p2_books})

q = "SELECT * FROM persons JOIN books ON name = author WHERE birthday < ?"
s1.execute_query(q, ["2001-01-01"], "result_s1")
s2.execute_query(q, ["2001-01-01"], "result_s2")
print("s1 局部结果 %.0f KB，s2 局部结果 %.0f KB" %
      (table_size(s1.get_local_data("result_s1")),
       table_size(s2.get_local_data("result_s2"))))

s1.send(s2, "result_s1")             # 只搬局部结果
s2.execute_query("SELECT * FROM result_s1 UNION SELECT * FROM result_s2",
                 [], "result_final")
print("最终结果行数:", s2.get_local_data("result_final").shape[0])
```

### 步骤 6：把观测到的数字与模型对照

把步骤 3/5 打印出的 KB 数与步骤 1 的模型值并列，回答：

1. 差多少？差异来自哪里（字符串逐行统计 vs 固定字节估算）？
2. 水平分区方案的传输量属于什么量级？为什么？

## 六、观测记录表

模型计算值（必做，把本机输出填入并注明与参照值是否一致）：

| 策略 | 计算式 | 参照值（KB） | 本机实测（KB） |
|---|---|---|---|
| ① 整表传输 | 10,000 × 52 B | 508 | |
| ② 广播连接 | 508 KB × 3 | 1,523 | |
| ③ 半连接归约 | 5,000 × 16 B + 10,000 × 0.6 × 52 B | 383 | |

选择率敏感性（半连接归约）：

| 选择率 | 0.1 | 0.2 | 0.4 | 0.6 | 0.8 | 1.0 |
|---|---|---|---|---|---|---|
| 传输量（KB） | | | | | | |
| 是否优于整表传输 | | | | | | |

节点数敏感性（广播连接）：

| 节点数 | 2 | 3 | 5 | 8 | 16 |
|---|---|---|---|---|---|
| 传输量（KB） | | | | | |

连接键宽度敏感性（半连接归约，选择率 0.6）：

| 键宽（B） | 4 | 8 | 16 | 32 |
|---|---|---|---|---|
| 传输量（KB） | | | | |

duckdb 复现（选做）：

| 项目 | 参照值（notebook） | 本机实测 |
|---|---|---|
| 整表传输量（KB） | 约 522 | |
| 连接键投影（KB） | 约 71 | |
| 过滤后的 books（KB） | 约 316 | |
| 半连接合计（KB） | 约 387 | |
| 水平分区：s1 局部结果（KB） | 需自行实测 | |
| 水平分区：s2 局部结果（KB） | 需自行实测 | |
| 三种做法的结果行数是否一致 | 必须一致 | |

## 七、思考题

1. 按本模型计算：整表传输、广播连接、半连接归约各传输多少 KB？请写出完整算式。
2. 半连接归约在什么情况下**不划算**？请用本模型求临界选择率（半连接代价 = 整表传输代价时的选择率）。
3. 把连接键从"姓名（16 B）"换成"整数 ID（4 B）"，半连接归约的传输量会降多少？为什么真实系统更爱用整数代理键做连接？
4. 广播连接的总传输量是整表传输的 3 倍（本实验 3 节点），为什么真实系统还是经常用它？
5. 水平协同分区为什么最省？如果两表**没有**按连接键分区，要付出什么代价？请用本模型估算一次 shuffle 的传输量。
6. 半连接归约要求"先发连接键投影"。如果过滤条件在 books 一侧（而不是 persons 一侧），这个做法还成立吗？方向该怎么选？
7. Spark 的 `spark.sql.autoBroadcastJoinThreshold` 默认 10485760（10 MB）是什么意思？表的估计大小超过它会发生什么？
8. Trino 的 `join-distribution-type = AUTOMATIC` 会在什么情况下选 BROADCAST、什么情况下选 PARTITIONED？没有统计信息时怎么办？

## 八、思考题答案与解析

**第 1 题**

```
① 整表传输 = 10,000 × 52 B = 520,000 B ÷ 1024 = 507.8 KB ≈ 508 KB
② 广播连接 = 507.8 KB × 3 = 1523.4 KB ≈ 1523 KB
③ 半连接   = (5,000 × 16 + 10,000 × 0.6 × 52) B
            = (80,000 + 312,000) B = 392,000 B ÷ 1024 = 382.8 KB ≈ 383 KB
```

**第 2 题**

半连接归约的代价随选择率 σ 线性上升：代价 = (80,000 + 520,000σ) B。令它等于整表传输 520,000 B：

```
80,000 + 520,000 σ = 520,000
σ = 440,000 / 520,000 ≈ 0.846
```

所以当**选择率高于约 85%**（即几乎所有人都要保留）时，半连接归约就不如直接整表传输——因为你发了一大把键，却几乎没过滤掉什么。

**第 3 题**

键宽从 16 B 降到 4 B：

```
投影 = 5,000 × 4 B = 20,000 B = 19.5 KB（原为 78 KB）
合计 = 20,000 + 312,000 = 332,000 B ÷ 1024 ≈ 324 KB（原为 383 KB）
```

降低约 59 KB（约 15%）。真实系统偏爱整数代理键（surrogate key）的原因：更窄（4–8 B vs 十几到几十字节）、定长（比较更快）、哈希更均匀、分区更均衡；而且用 ID 做连接键时，姓名变更不会波及外键。

**第 4 题**

广播连接虽然总传输量是 3 倍，但它换来的是**并行度**：每个节点都拿到了小表，可以各自对自己那份大表做本地连接，三个节点同时算。整表传输则把工作全压到一个节点上，其他节点空转。当小表足够小（Spark 默认阈值 10 MB 以内）时，"多传一点、换全部节点并行"通常是划算的。反过来说，节点越多，广播的总量越大，节点数很大时就该改用哈希重分布。

**第 5 题**

协同分区时，能匹配上的记录一定落在同一个节点，所以连接是纯本地操作，网络只需要搬运**最终结果**（往往远小于输入）。

若两表没有按连接键分区，就必须先做 **shuffle**（重分布）：按连接键的哈希值把每张表的记录发到对应节点。按本模型估算，两张表都要搬：

```
persons: 5,000 × 60 B = 300,000 B ≈ 293 KB
books  : 10,000 × 52 B = 520,000 B ≈ 508 KB
合计                              ≈ 801 KB
```

这比半连接归约（383 KB）还贵一倍以上——这就是"分区方式决定连接代价"的直接体现。

**第 6 题**

成立，但要**选对方向**：半连接归约的方向应该从"过滤后更小的一侧"出发，把它的连接键投影发给另一侧做过滤。若过滤条件在 books 一侧（例如 `WHERE price < 20`），则应该：

1. s2 本地过滤 books，取出 `DISTINCT author`；
2. 把 author 投影发给 s1；
3. s1 用它半连接过滤 persons；
4. 把过滤后的 persons 发回 s2（或反过来）；
5. 在某一侧完成最终连接。

一般原则：**先算选择率，谁小谁先投影**。如果两侧都有过滤条件，就都先本地过滤再投影。

**第 7 题**

`spark.sql.autoBroadcastJoinThreshold` 是允许广播的表的大小上限，默认 10485760 字节 = 10 MB。优化器会读表的统计信息估计大小：估计值 ≤ 10 MB 就走 broadcast join（把小表发到每个 executor）；超过阈值就改用 Sort-Merge Join，两侧都按连接键 shuffle 重分布。把它设为 −1 可以彻底关闭广播。另外，Spark 3.2.0 起 AQE（Adaptive Query Execution）默认开启，它能在运行时根据实际统计把 Shuffle Join 动态转成 Broadcast Join。

**第 8 题**

Trino 的 `join-distribution-type = AUTOMATIC` 的逻辑：

- **BROADCAST**：右表（被广播的一侧）过滤后足够小，能放进每个 worker 的内存 → 广播右表，左表不动，各 worker 本地连接；
- **PARTITIONED**：两侧都大 → 按连接键哈希重分布两侧，再做分布式连接；
- **没有统计信息时**：优化器无法判断表大小，回退到哈希分布式（PARTITIONED），这是一种保守选择——它不会因误判而撑爆内存。

这也是为什么"收集统计信息（ANALYZE）"在分布式系统里格外重要：没有它，优化器只能选保守策略。

## 九、扩展任务（选做）

1. **双向过滤**：给 books 也加一个过滤条件（如 `price < 20`），实现"两侧都先过滤再投影"的半连接，测传输量。
2. **多级半连接**：三张表连接时，设计"先投影两次再传数据"的方案，与两表情形比较。
3. **代价模型扩展**：把"网络传输量"换成"传输量 / 带宽 + 各节点 CPU 时间"，考虑并行度，重新比较三种策略。
4. **倾斜（skew）实验**：让某个作者的书籍特别多（例如某作者占 30% 的 books），观察协同分区后各节点的负载是否均衡，思考倾斜怎么处理。
5. **真实引擎对照**：如果有 Spark 或 Trino 环境，跑同一条查询，用 `EXPLAIN` 看优化器选了哪种分布方式，与本实验的结论对照。

## 十、常见错误与排查

| 现象 | 原因 | 处理 |
|---|---|---|
| 半连接归约算出来比整表传输还贵 | 选择率取了 1.0（没有过滤效果），或键宽设得过大 | 检查选择率；键宽按实际列类型填（姓名约 16 B，整数 ID 4 B） |
| 三种做法的结果行数不一致 | 过滤条件只在一侧生效，或 `DISTINCT` 投影丢了重复键导致多匹配 | 半连接只用于过滤，最终连接仍要用完整条件；结果集用 `sort_values` 后比对 |
| `assert_frame_equal` 报列顺序不同 | 不同做法的 SELECT 列顺序不同（如 `books.*` vs `*`） | 比对前统一列顺序：`df = df[df.columns.sort_values()]` |
| 水平分区后出现重复或缺失记录 | 分区谓词用了 `<` 与 `>`，漏掉了等于边界 | 用 `< "M"` 与 `>= "M"` 这种互补的划分 |
| `table_size()` 慢得离谱 | 用 `iterrows()` 逐行遍历大表 | 本实验数据量下可接受；数据量大时改成向量化的 `df.applymap` 或按列估算 |
| duckdb 报 `Table with name xxx does not exist` | 只在 `local_data` 里存了，没有 `con.register()` | `receive()` 里必须 `register`，duckdb 才能查到 |
| 广播连接随节点数线性增长，看起来很亏 | 这是模型本身的结果，不是 bug | 记住广播换来的是并行度，代价与收益要一起看 |
| `ModuleNotFoundError: No module named 'duckdb'` | 依赖没装 | 只做步骤 1、2、5，或 `pip install duckdb faker` |
| faker 生成的数据每次不同 | 没设种子 | 加 `Faker.seed(42)`，保证可复现 |