{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-0",
"source": [
"# 实验 15：分布式连接\n",
"\n",
"## 为什么做这个实验\n",
"\n",
"数据不在一台机器上时，连接的玩法会整个变掉。\n",
"\n",
"单机里优化查询，第一目标是\"算得快\"——减少 CPU 与磁盘 I/O；分布式里，瓶颈往往从 CPU 变成了**网络**，\n",
"第一目标也随之变成\"**少搬数据**\"。局域网带宽通常比本地磁盘顺序读慢一到两个数量级，\n",
"搬错一次数据，省下的 CPU 时间连零头都抵不回来。\n",
"\n",
"于是\"怎么把两张分处不同节点的表连起来\"就有了多种方案，它们的**结果完全相同**，\n",
"差别只在搬了多少字节：\n",
"\n",
"| 策略 | 做法 | 什么时候划算 |\n",
"| --- | --- | --- |\n",
"| 整表传输 | 把一张表整个搬到另一张表所在的节点 | 表小，或者对方节点算力冗余 |\n",
"| 广播连接 | 把小表复制到**每一个**有大表的节点 | 小表足够小，且想保住并行度 |\n",
"| 半连接归约 | 先发连接键投影去过滤，再发过滤后的数据 | 过滤条件选择率低、连接键投影远小于整行 |\n",
"\n",
"本实验用内存版 DuckDB 模拟三个节点，把这三种方案的传输量**量出来**并对比，\n",
"最后用单元测试证明一件事：**不同执行策略的结果集必须一模一样**——差的只是代价，不是答案。\n",
"\n",
"## 运行前准备\n",
"\n",
"- Python 3.9 及以上。\n",
"- 依赖：`duckdb`（每个节点一个内存实例）、`pandas`（搬的数据就是 DataFrame）、`faker`（造演示数据）。\n",
"- 缺包时安装：\n",
"\n",
" ```bash\n",
" python -m pip install duckdb pandas faker\n",
" ```\n",
"\n",
" （本机已配置环境可用：`python -m pip install duckdb pandas faker`）\n",
"\n",
"- 打开方式：在仓库根目录执行 `jupyter notebook notebooks/Distributed-Joins.ipynb`。\n",
" 第一个代码 cell 会把仓库根目录加进 `sys.path`，这样 `from system.data_classes import *` 才找得到 `Person` 与 `Book`。\n",
"- 若实在装不上 `duckdb`，本实验的核心（三种策略的传输量模型）仍可用纯 Python 算，见末尾思考题。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-1",
"source": [
"## 要解决的问题\n",
"\n",
"```sql\n",
"SELECT * \n",
"FROM persons JOIN books\n",
" ON name = author\n",
"WHERE birthday < 2001-01-01\n",
"```\n",
"\n",
"两个前提要盯住，因为它们直接决定搬多少数据：\n",
"\n",
"1. **数据怎么分布**（见下面两种场景）：\n",
" - 垂直分区：每个节点存**整张**表，不同表在不同节点；\n",
" - 水平分区：每张表按某个键切成若干份，分散到各节点。\n",
"2. **过滤条件在哪一侧**：`birthday < 2001-01-01` 作用在 `persons` 上。\n",
" 它决定了半连接归约应该从哪一侧先发连接键投影。\n",
"\n",
"下面先造数据、定义\"节点\"，然后分别在两种分布场景下执行这条查询。"
]
},
{
"cell_type": "code",
"id": "2927ee85-116b-47a4-9b0c-66609bc730ef",
"metadata": {},
"source": [
"from __future__ import (\n",
" annotations,\n",
") # postponed evaluation of annotations https://peps.python.org/pep-0563/\n",
"\n",
"import sys, pathlib\n",
"\n",
"# Put the repo root (the directory containing `system/`) on sys.path so\n",
"# `from system...` works whether this notebook is opened from its own\n",
"# folder or from the repo root.\n",
"for _p in (pathlib.Path.cwd(), *pathlib.Path.cwd().parents):\n",
" if (_p / \"system\").is_dir():\n",
" if str(_p) not in sys.path:\n",
" sys.path.insert(0, str(_p))\n",
" break\n",
"\n",
"import duckdb\n",
"import pandas as pd\n",
"from typing import Any\n",
"from faker import Faker\n",
"from system.data_classes import *"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "code",
"id": "1455827f-3e9d-4b46-8811-1e891bb2a46c",
"metadata": {},
"source": [
"Faker.seed(42)\n",
"fake = Faker()\n",
"\n",
"number_of_tuples = 10000\n",
"\n",
"# create some fake data mimicking tuples in a relation\n",
"persons = [Person(fake.name(), fake.date()) for _ in range(number_of_tuples)]\n",
"books = [\n",
" Book(\n",
" fake.sentence(),\n",
" persons[i % number_of_tuples].name,\n",
" fake.pyfloat(min_value=0, max_value=100, right_digits=2),\n",
" )\n",
" for i in range(number_of_tuples)\n",
"]\n",
"\n",
"# convert to Pandas DataFrame\n",
"persons_df = pd.DataFrame(persons)\n",
"books_df = pd.DataFrame(books)\n",
"display(persons_df[:5])\n",
"display(books_df[:5])"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-4",
"source": [
"## 定义\"节点\"（Server）\n",
"\n",
"`Server` 类把一个节点抽象成两样东西：\n",
"\n",
"- `con`：该节点上跑着的 DBMS 实例（这里是一个内存 DuckDB 连接）；\n",
"- `local_data`：该节点的本地数据区，字典形式，名字 → DataFrame。\n",
"\n",
"三个关键方法：\n",
"\n",
"| 方法 | 作用 |\n",
"| --- | --- |\n",
"| `import_data` | 把表导入本节点的 DBMS |\n",
"| `execute_query` | 在本节点执行查询，结果存进本地数据区并**注册**回 DuckDB，供后续查询引用 |\n",
"| `send` / `receive` | 模拟网络传输：`send` 取出本地数据交给对方，`receive` 把它注册进对方的 DuckDB |\n",
"\n",
"请注意这是**模拟**，不是真的网络：`send` 只是把同一个 DataFrame 的引用交给另一个对象，\n",
"没有序列化、没有带宽限制。所以本实验能观测的是\"**搬了多少数据**\"，而不是\"搬了多久\"。\n",
"要测真实耗时，得用真正的分布式引擎（Spark / Trino）。"
]
},
{
"cell_type": "code",
"id": "63fc7da40b0654b6",
"metadata": {},
"source": [
"class Server:\n",
" \"\"\"A simple server that contains a duckdb connection.\n",
"\n",
" Attributes:\n",
" con (duckdb.duckdb.DuckDBPyConnection): The duckdb connection.\n",
" local_data (dict[str, pd.DataFrame): The local data.\n",
" \"\"\"\n",
"\n",
" def __init__(self, con: duckdb.duckdb.DuckDBPyConnection):\n",
" self.con: duckdb.duckdb.DuckDBPyConnection = con # DBMS running on this server\n",
" self.local_data: dict[str, pd.DataFrame] = {}\n",
"\n",
" def import_data(self, tables: dict[str, pd.DataFrame]):\n",
" \"\"\"Import the data tables into the duckdb instance.\n",
"\n",
" :param tables: A dictionary of table name and data to import.\n",
" \"\"\"\n",
" # duckdb resolves the literal `df_name` in the query below by the\n",
" # Python variable name (replacement scan), so df_name is used even\n",
" # though it never appears interpolated.\n",
" # pylint: disable-next=unused-variable\n",
" for table_name, df_name in tables.items():\n",
" self.con.execute(f\"CREATE TABLE {table_name} AS SELECT * FROM df_name\")\n",
"\n",
" def execute_query(self, query: str, parameters: list[Any], name: str = \"result\"):\n",
" \"\"\"Execute the given `query` with its `parameters` on the duckdb instance. The result is stored in the local data area with the name `name` and registered in the duckdb instance.\n",
"\n",
" :param query: The query to execute.\n",
" :param parameters: The parameters belonging to the query. Required to execute the query as a prepared statement.\n",
" :param name: The name of the result table.\n",
" \"\"\"\n",
" result_df = self.con.execute(query, parameters).fetchdf() # prepared statement\n",
" self.local_data[name] = result_df\n",
" self.con.register(name, result_df)\n",
"\n",
" def get_local_data(self, name: str):\n",
" \"\"\"Retrieve the data table specified by the name `name` from the local data area.\n",
"\n",
" :param name: The name of the data table to retrieve.\n",
" :return: The data table specified by the name `name`.\n",
" \"\"\"\n",
" try:\n",
" data_df = self.local_data[name]\n",
" return data_df\n",
" except KeyError:\n",
" print(f\"No local data for {name}.\")\n",
" return\n",
"\n",
" def receive(self, name: str, data_df: pd.DataFrame):\n",
" \"\"\"Receive a table with the name `name` and the data given by `data_df`. Simulates receiving data by registering the data in the duckdb instance.\n",
"\n",
" :param name: The name of the data table.\n",
" :param data_df: The data to be received.\n",
" \"\"\"\n",
" self.con.register(\n",
" name, data_df\n",
" ) # simulate receiving data by registering the data in the duckdb instance\n",
"\n",
" def send(self, other: Server, local_data_name: str):\n",
" \"\"\"Sends the data specified by `local_data_name` from the servers local data area to the `other` server.\n",
"\n",
" :param other: Server to send the data to.\n",
" :param local_data_name: The name of the local data table to send.\n",
" \"\"\"\n",
" data_df = self.get_local_data(local_data_name)\n",
" other.receive(\n",
" local_data_name, data_df\n",
" ) # simulate sending the data to another server"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-6",
"source": [
"## 怎么度量\"搬了多少数据\"\n",
"\n",
"`table_size()` 的口径要记牢，后面所有数字都依赖它：\n",
"\n",
"- 字符串按**实际字符长度**计字节；\n",
"- 数值等非字符串一律按 **4 字节**计；\n",
"- 最后除以 1024 换算成 KB。\n",
"\n",
"这是一个**近似**估算，不是 DuckDB 的真实存储占用。另有一套口径是\"条数 × 每条固定字节数\"\n",
"（本实验的模型值：persons 60 B/条、books 52 B/条）。两套口径都可用，\n",
"但**比较时必须统一口径**，否则会得出错误的结论。"
]
},
{
"metadata": {},
"cell_type": "code",
"source": [
"# Utility function to compute the size of a database table\n",
"def table_size(table: pd.DataFrame):\n",
" \"\"\"Returns the size of a pandas DataFrame in kilobytes. This function assumes 4 byte for numeric/non-string values and counts the length of the string otherwise.\n",
"\n",
" :param table: pandas DataFrame representing a table.\n",
" :return: size of the table in kilobytes.\n",
" \"\"\"\n",
" size = 0 # byte\n",
" for _, row in table.iterrows():\n",
" for e in row:\n",
" if type(e) is str:\n",
" size += len(e) # length of string\n",
" else:\n",
" size += 4 # assume 4 byte for numeric/non-string values\n",
" return size / 1024"
],
"id": "1b117253-c6c7-452f-93b1-f588925648ae",
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-7",
"source": [
"## 场景 1：垂直分区——每个节点存一整张表\n",
"\n",
" s1 s2\n",
"┌──────────────┐ ┌──────────────┐\n",
"│ persons │ │ books │\n",
"│ (整张表) │ │ (整张表) │\n",
"└──────────────┘ └──────────────┘\n",
" └────────── 网络 ────────┘\n",
"\n",
"这是最\"直白\"的分布：一张表整个待在一个节点上。此时要做连接，就必须把某一侧搬过去\n",
"（或者把两侧都搬到第三个节点）。下面比较两种搬法。"
]
},
{
"cell_type": "code",
"id": "693c68fd-71bb-4fb7-89cb-9164df24b918",
"metadata": {},
"source": [
"# Setup server 1\n",
"s1 = Server(duckdb.connect(database=\":memory:\"))\n",
"s1.import_data({\"persons\": persons_df}) # s1 contains entire Persons table\n",
"\n",
"# Setup server 2\n",
"s2 = Server(duckdb.connect(database=\":memory:\"))\n",
"s2.import_data({\"books\": books_df}) # s2 contains entire Books table"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-9",
"source": [
"## 做法 1.1：整表传输\n",
"\n",
"最直接的做法：把 `books` 整表从 s2 搬到 s1，然后在 s1 上完成全部工作。\n",
"\n",
"执行步骤：\n",
"\n",
"1. s2 取出整张 `books`，放进自己的本地数据区（命名 `books_local`）；\n",
"2. s2 把它 `send` 给 s1；\n",
"3. s1 在本地做 `persons ⋈ books_local`，再套上 `birthday < ?` 过滤，得到最终结果。\n",
"\n",
"**代价**：整张 books 的字节数。\n",
"**问题**：① 数据里可能有一大半是无用的（作者不满足生日条件的那些书），全搬了；\n",
"② 工作全压在 s1 上，s2 搬完就闲着——**并行度丢了**。\n",
"\n",
"预期看到：`Sending books table of size ... KB.` 后面跟着结果行数。\n",
"把这个 KB 数记下来，它是后面两种做法的对照基线。"
]
},
{
"cell_type": "code",
"id": "760d09ae-8151-46c9-b5d4-6b371297df78",
"metadata": {},
"source": [
"# 1. Extract Books table on s2 and store it in the local data area of s2 with the name \"books_local\"\n",
"s2.execute_query(\"SELECT * FROM books\", [], \"books_local\")\n",
"\n",
"# 2. Send copy of the Books table to s1\n",
"s2.send(s1, \"books_local\")\n",
"print(f\"Sending books table of size {table_size(s2.get_local_data(\"books_local\"))} KB.\")\n",
"\n",
"# 3. s1 now contains the Books data in its local data area -> perform join between Persons and Books\n",
"s1.execute_query(\n",
" \"\"\"SELECT *\n",
" FROM persons JOIN books_local on name = author\n",
" WHERE birthday < ?\n",
" \"\"\",\n",
" [\"2001-01-01\"],\n",
" \"result\", # name used to store the result in the local data area of s1\n",
")\n",
"result_entire_table = s1.get_local_data(\n",
" \"result\"\n",
") # retrieve the previously computed result\n",
"print(\"Number of results:\", result_entire_table.shape[0])"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-11",
"source": [
"## 做法 1.2：半连接归约（semi-join reduction）\n",
"\n",
"上一个做法的问题在于\"搬了不该搬的\"。改进思路：既然只关心 21 世纪前出生的人写的书，\n",
"那就**先把这个信息（连接键投影）发过去做过滤，再搬过滤后的数据**。\n",
"\n",
"术语：**半连接**（semi-join）`R ⋉ S` 只保留 R 中\"在 S 里能找到匹配\"的记录，\n",
"不重复、也不带 S 的列。它在这里的用途是**过滤**，不是产出结果。\n",
"\n",
"执行步骤：\n",
"\n",
"1. s1 本地执行 `SELECT DISTINCT name FROM persons WHERE birthday < ?`，取出满足条件的连接键投影；\n",
"2. s1 把这份投影发给 s2——它只有一列姓名，**远小于整张 persons**；\n",
"3. s2 用它做半连接 `books ⋈ 这些键`，本地过滤出合格的 books；\n",
"4. s2 把过滤后的 books 发给 s1；\n",
"5. s1 用完整的过滤条件完成最终连接。\n",
"\n",
"注意两点：\n",
"\n",
"- 第 1 步用了 `DISTINCT`：投影里同一个键出现多次没有意义，去重能再省一点；\n",
"- 第 5 步**仍然要再写一遍** `persons.birthday < ?`。半连接只用于减少传输，\n",
" 最终结果的正确性还得靠完整条件保证。\n",
"\n",
"预期看到两段发送日志：一段是投影（很小），一段是过滤后的 books（明显小于整表）。"
]
},
{
"cell_type": "code",
"id": "74cfb406-899f-4330-b304-4386a8163d98",
"metadata": {},
"source": [
"# 1. Extract the primary key of all persons born before the 21th century on s1 and store the result in the local data area of s1 with the name \"persons_filtered\"\n",
"s1.execute_query(\n",
" \"SELECT DISTINCT name FROM persons WHERE birthday < ?\",\n",
" [\"2001-01-01\"],\n",
" \"persons_filtered\",\n",
")\n",
"\n",
"# 2. Send copy of filtered persons to S2\n",
"s1.send(s2, \"persons_filtered\")\n",
"persons_filtered_size = table_size(s1.get_local_data(\"persons_filtered\"))\n",
"print(f\"Sending filtered persons table of size {persons_filtered_size} KB.\")\n",
"\n",
"# 3. Extract books written by persons born before the 21th century\n",
"s2.execute_query(\n",
" \"SELECT books.* FROM books JOIN persons_filtered ON author = name\",\n",
" [],\n",
" \"books_filtered\",\n",
") # semi-join\n",
"\n",
"# 4. Send copy of filtered books to s1\n",
"s2.send(s1, \"books_filtered\")\n",
"books_filtered_size = table_size(s2.get_local_data(\"books_filtered\"))\n",
"print(f\"Sending filtered books table of size {books_filtered_size} KB.\")\n",
"\n",
"# 5. Execute final query, note that we could also reuse the pre-filtered persons table\n",
"s1.execute_query(\n",
" \"\"\"SELECT *\n",
" FROM persons JOIN books_filtered on name = author\n",
" WHERE persons.birthday < ?\n",
" \"\"\",\n",
" [\"2001-01-01\"],\n",
" \"result\",\n",
")\n",
"result_semi_join = s1.get_local_data(\n",
" \"result\"\n",
") # retrieve the previously computed result\n",
"print(\"Number of results:\", result_semi_join.shape[0])"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-13",
"source": [
"## 对比：搬了多少数据\n",
"\n",
"仓库 notebook 的实测输出（`table_size()` 按字符串实际长度逐行统计）：\n",
"\n",
"- 做法 1.1 整表传输：约 **522 KB**；\n",
"- 做法 1.2 半连接归约：约 71 KB（连接键投影）+ 316 KB（过滤后的 books）= **387 KB**。\n",
"\n",
"另有一套\"固定字节估算\"的模型口径（persons 5 000 条 × 60 B、books 10 000 条 × 52 B、\n",
"3 个节点、过滤选择率 60%、连接键 16 B）：\n",
"\n",
"| 策略 | 算式 | 传输量 |\n",
"| --- | --- | --- |\n",
"| ① 整表传输 | 10 000 × 52 B | **508 KB** |\n",
"| ② 广播连接 | 508 KB × 3 | **1 523 KB** |\n",
"| ③ 半连接归约 | 5 000 × 16 B + 10 000 × 0.6 × 52 B | **383 KB** |\n",
"\n",
"两套数字量级一致，差异来自口径不同（逐行按字符串实际长度 vs 每条按固定字节估算）。\n",
"**比较时必须统一口径**，不要拿 522 去比 383。\n",
"\n",
"按模型口径，半连接归约比整表传输少约 25%，比广播连接少约 75%。\n",
"\n",
"还有一句代码注释里的提醒值得展开：这里的连接键是**姓名**（`name = author`），不是整数 ID。\n",
"姓名又长又不均匀，如果换成整数代理键（4 B），第 1 步的投影会从 5 000 × 16 B 降到 5 000 × 4 B，\n",
"半连接归约还能再省一截——这也是真实系统偏爱整数代理键做连接的原因之一。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-14",
"source": [
"## 场景 2：水平分区——每个节点存每张表的一部分\n",
"\n",
" s1 s2\n",
"┌────────────────────┐ ┌────────────────────┐\n",
"│ persons (A–L) │ │ persons (M–Z) │\n",
"│ books (A–L) │ │ books (M–Z) │\n",
"└────────────────────┘ └────────────────────┘\n",
" 按连接键（姓名）的姓氏首字母切分 → 协同分区\n",
"\n",
"**协同分区**（co-partitioning）：两张表按**同一个连接键**、用**同一个分区函数**切分。\n",
"这样一来\"能连上的记录一定落在同一个节点\"，连接可以在各节点**本地**独立完成，\n",
"网络只需要搬运最终的（或部分）结果。\n",
"\n",
"关键点：切分键必须**就是连接键**。如果两表没有按连接键分区，就必须先 **shuffle**（重分布）——\n",
"按连接键的哈希值把每张表的记录发到对应节点，那通常是最贵的一种做法。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-16",
"source": [
"## 分区函数：按连接键切分\n",
"\n",
"`partition()` 取每条记录的姓名（persons 用 `name`，books 用 `author`），\n",
"按**空格拆分取第 2 段**（即姓氏）的首字母，以 'M' 为界切成两份：\n",
"\n",
"- 姓氏首字母 < 'M' → 分区 1（放 s1）；\n",
"- 否则 → 分区 2（放 s2）。\n",
"\n",
"两个细节值得注意：\n",
"\n",
"1. **两表必须用同一个函数、同一个键切分**。persons 按 `name`、books 按 `author`，\n",
" 而连接条件正是 `name = author`，所以同一个人的记录必然落在同一个节点——这就是协同分区。\n",
"2. **切分谓词必须互补**。这里用 `< 'M'` 与其否定 `>= 'M'`（代码里是 `else` 分支），\n",
" 边界上一个不漏、一个不重。若写成 `< 'M'` 与 `> 'M'`，恰好姓 'M' 开头的人会被两边同时丢掉。\n",
"\n",
"这种按首字母切分的方式有个明显缺点：**姓氏分布不均匀**，S 开头的可能比 Q 开头的多好几倍，\n",
"两个节点负载失衡。真实系统用哈希分区来缓解这一点，但哈希也救不了\"某个键本身特别密集\"的倾斜。"
]
},
{
"cell_type": "code",
"id": "d75408ef-3c61-4e39-91d5-eb0cdfb62bc4",
"metadata": {},
"source": [
"def partition(data: list, attribute_name: str):\n",
" \"\"\"Partitions the input data based on the first character of the last name.\n",
" :param data: List of entities to partition.\n",
" :param attribute_name: Name of the attribute on which to partition the data.\n",
" :return: Two pandas.DataFrames containing both partitions.\n",
" \"\"\"\n",
" p1 = []\n",
" p2 = []\n",
" for e in data:\n",
" last_name = getattr(e, attribute_name).split(\" \")[1]\n",
" if last_name[0] < \"M\":\n",
" p1.append(e)\n",
" else:\n",
" p2.append(e)\n",
" return pd.DataFrame(p1), pd.DataFrame(p2)"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "code",
"id": "6a20112a-5a4e-42be-9b38-2174006ce150",
"metadata": {},
"source": [
"# Partition the data based on the join attribute\n",
"s1_persons_df, s2_persons_df = partition(persons, \"name\")\n",
"s1_books_df, s2_books_df = partition(books, \"author\")\n",
"\n",
"# Setup server 1\n",
"s1 = Server(duckdb.connect(database=\":memory:\"))\n",
"s1.import_data(\n",
" {\"persons\": s1_persons_df, \"books\": s1_books_df}\n",
") # s1 contains a partition of Books and Persons\n",
"\n",
"# Setup server 2\n",
"s2 = Server(duckdb.connect(database=\":memory:\"))\n",
"s2.import_data(\n",
" {\"persons\": s2_persons_df, \"books\": s2_books_df}\n",
") # s2 contains a partition of Books and Persons"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-17",
"source": [
"## 做法 2.1：本地连接\n",
"\n",
"因为两表已经按连接键协同分区，\"能匹配的记录必在同一个节点\"，所以每个节点可以各自算自己的那份：\n",
"\n",
"1. s1 算姓氏首字母 < 'M' 的那份局部连接结果，s2 算 ≥ 'M' 的那份；\n",
"2. s1 把自己的局部结果发给 s2（代码里挑了较小的那份来发）；\n",
"3. s2 用 `UNION` 合并两份局部结果，得到最终结果。\n",
"\n",
"**代价**：网络只搬了**局部结果**，而不是任何一张输入表。这通常是最省的一种做法——\n",
"代价从\"搬输入\"变成了\"搬输出\"。\n",
"\n",
"代价转移到了别处：它要求数据**事先**按连接键分区好。分区本身是一次全量重排，\n",
"而且如果某个键特别密集（数据倾斜），对应节点会明显更慢。"
]
},
{
"cell_type": "code",
"id": "52aa48e8-d021-4c95-85e9-667b0396a06b",
"metadata": {},
"source": [
"# 1. Compute the partial join result locally on each node\n",
"s1.execute_query(\n",
" \"SELECT * FROM persons JOIN books ON name = author WHERE birthday < ?\",\n",
" [\"2001-01-01\"],\n",
" \"result_s1\",\n",
")\n",
"s2.execute_query(\n",
" \"SELECT * FROM persons JOIN books ON name = author WHERE birthday < ?\",\n",
" [\"2001-01-01\"],\n",
" \"result_s2\",\n",
")\n",
"\n",
"# 2. Send `result_s1` to s2 since it is smaller than `result_s2`\n",
"s1.send(s2, \"result_s1\")\n",
"result_s1_size = table_size(s1.get_local_data(\"result_s1\"))\n",
"print(f\"Sending partial result computed on s1 of size {result_s1_size} KB to s2.\")\n",
"\n",
"# Compute final result by combining both partial results\n",
"s2.execute_query(\n",
" \"SELECT * FROM result_s1 UNION SELECT * FROM result_s2\", [], \"result_final\"\n",
")\n",
"result_local_joins = s2.get_local_data(\"result_final\")\n",
"print(\"Number of results:\", result_local_joins.shape[0])"
],
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-19",
"source": [
"## 单元测试：三种做法必须给出同一个答案\n",
"\n",
"这是本实验最重要的一条断言：整表传输、半连接归约、本地连接三种执行策略，\n",
"**结果集必须完全相同**。它们只是代价不同，不是语义不同。\n",
"\n",
"比对方式：三种结果的**行顺序**可能不同（不同执行计划产出顺序不同），\n",
"所以先把每个 DataFrame 按**全部列**排序并重置索引，再用 `assert_frame_equal` 逐格比对。\n",
"\n",
"如果这条断言挂了，说明某个环节写错了——最常见的两种：\n",
"\n",
"- 半连接那一步把 `DISTINCT` 投影当成了最终结果，丢掉了重复匹配；\n",
"- 水平分区的切分谓词不互补（例如一边用 `<` 另一边用 `>`），边界记录被漏掉。\n",
"\n",
"跑完应看到 `Ran 1 test ... OK`。"
]
},
{
"metadata": {},
"cell_type": "code",
"source": [
"import unittest\n",
"from pandas.testing import assert_frame_equal\n",
"\n",
"\n",
"class MyTest(unittest.TestCase):\n",
"\n",
" def test_result_sets(self):\n",
" # Compare the three previously computed result sets.\n",
" # Sort each table on all columns to ensure that the order of rows match\n",
" result_11 = result_entire_table.sort_values(\n",
" by=result_entire_table.columns.tolist()\n",
" ).reset_index(\n",
" drop=True\n",
" ) # result computed by approach 1.1\n",
" result_12 = result_semi_join.sort_values(\n",
" by=result_semi_join.columns.tolist()\n",
" ).reset_index(\n",
" drop=True\n",
" ) # result computed by approach 1.2\n",
" result_21 = result_local_joins.sort_values(\n",
" by=result_local_joins.columns.tolist()\n",
" ).reset_index(\n",
" drop=True\n",
" ) # result computed by approach 2.1\n",
"\n",
" assert_frame_equal(result_11, result_12)\n",
" assert_frame_equal(result_12, result_21)\n",
"\n",
"\n",
"# Run unit tests\n",
"unittest.main(argv=[\"ignored\", \"-v\"], defaultTest=\"MyTest\", verbosity=2, exit=False)"
],
"id": "731629bee1b9848",
"outputs": [],
"execution_count": null
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-90",
"source": [
"## 真实引擎里的对应物\n",
"\n",
"本 notebook 手工实现了三种策略；真实系统把这些选择做进了优化器：\n",
"\n",
"- **Spark**：`spark.sql.autoBroadcastJoinThreshold` 默认 10485760 字节（10 MB）。\n",
" 优化器估计某张表的大小 ≤ 10 MB 就走 broadcast join（把小表发到每个 executor，各自本地连接），\n",
" 超过就改用 Sort-Merge Join，两侧都按连接键 shuffle 重分布。设为 −1 可彻底关闭广播。\n",
" 另外 **AQE**（Adaptive Query Execution，自适应查询执行）自 **Spark 3.2.0** 起默认开启，\n",
" 它能在运行时根据实际统计把 Shuffle Join 动态转成 Broadcast Join——这正是\"估计不准就补救\"的思路。\n",
"- **Trino**：`join-distribution-type` 默认 `AUTOMATIC`。右表过滤后足够小 → BROADCAST；\n",
" 两侧都大 → PARTITIONED（哈希重分布两侧）；**没有统计信息时回退到 PARTITIONED**\n",
" （保守选择，不会因误判而撑爆内存）。\n",
"- **GIS 场景**：如果连接涉及空间谓词（例如\"哪些订单落在某个配送区域内\"），\n",
" 连接键不再是标量而是几何对象，分区与索引都要另做。Apache Sedona 在 Spark 之上提供了\n",
" 空间分区、空间索引与空间连接（spatial join）算子，是这类负载的常用迁移路径。\n",
"\n",
"一句话总结：**本实验手工做的\"选哪种搬法\"，在真实引擎里是优化器根据统计信息自动做的。**\n",
"所以收集统计信息（ANALYZE）在分布式系统里格外重要。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"id": "gp10-15-91",
"source": [
"## 思考题与详解\n",
"\n",
"**1. 按\"固定字节估算\"的模型口径计算：整表传输、广播连接、半连接归约各传输多少 KB？写出完整算式。**\n",
"\n",
"答案要点
\n",
"\n",
"已知：persons 5 000 条 × 60 B，books 10 000 条 × 52 B，3 个节点，\n",
"过滤选择率 σ = 0.6，连接键宽 16 B，1 KB = 1024 B。\n",
"\n",
"- ① 整表传输 = 10 000 × 52 B = 520 000 B ÷ 1024 = **507.8 KB ≈ 508 KB**\n",
"- ② 广播连接 = 507.8 KB × 3 = **1 523.4 KB ≈ 1 523 KB**\n",
"- ③ 半连接归约 = (5 000 × 16 + 10 000 × 0.6 × 52) B = (80 000 + 312 000) B = 392 000 B ÷ 1024 = **382.8 KB ≈ 383 KB**\n",
"\n",
"半连接 / 整表 ≈ 0.75，半连接 / 广播 ≈ 0.25。\n",
"\n",
"答案要点
\n",
"\n",
"半连接代价随选择率 σ **线性上升**：代价(σ) = (80 000 + 520 000 σ) B。\n",
"令它等于整表传输的 520 000 B：\n",
"\n",
" 80 000 + 520 000 σ = 520 000 ⇒ σ = 440 000 / 520 000 ≈ 0.846\n",
"\n",
"所以当**选择率高于约 85%** 时，半连接归约不如直接整表传输——你发了一大把连接键，\n",
"却几乎没过滤掉什么，等于多付了一次投影的钱。\n",
"\n",
"直觉：半连接赚的是\"过滤掉的那一小部分\"。σ 越接近 1，能过滤掉的越少，赚头越小，\n",
"而发键的固定开销（80 000 B）一分不少。\n",
"\n",
"答案要点
\n",
"\n",
"- 投影：5 000 × 4 B = 20 000 B（原为 80 000 B），省 60 000 B；\n",
"- 合计：20 000 + 312 000 = 332 000 B ÷ 1024 ≈ **324 KB**（原为 383 KB），降低约 59 KB，约 **15%**。\n",
"\n",
"（注：本实验里降低幅度不算大，是因为过滤后的 books 那 312 000 B 占了大头。\n",
"真实场景里连接键投影占比更高时，收益会明显得多。）\n",
"\n",
"真实系统偏爱整数代理键（surrogate key）的原因：\n",
"\n",
"- **更窄**：4–8 B，而姓名通常十几到几十字节；\n",
"- **定长**：比较与哈希更快，分区更均衡；\n",
"- **稳定**：用 ID 做外键时，姓名变更不会波及引用它的表；\n",
"- **哈希均匀**：整数哈希分布好，shuffle 时不易倾斜。\n",
"\n",
"答案要点
\n",
"\n",
"因为它换来的是**并行度**。\n",
"\n",
"- 广播：每个节点都拿到小表，可以各自对自己那份大表做本地连接，**N 个节点同时算**；\n",
"- 整表传输：把工作全压到一个节点，其他节点搬完就空转。\n",
"\n",
"当小表足够小（Spark 默认阈值 10 MB 以内）时，\"多传一点、换全部节点并行\"通常是划算的——\n",
"网络多传 2 份的代价，远小于让 2/3 的节点闲着。\n",
"\n",
"反过来说：**节点越多，广播的总量越大**（线性增长），节点数很大时就该改用哈希重分布。\n",
"所以广播不是\"永远好\"，而是\"小表 + 中等规模集群\"下的好选择。\n",
"\n",
"答案要点
\n",
"\n",
"协同分区时，能匹配上的记录**一定落在同一个节点**，所以连接是纯本地操作，\n",
"网络只需要搬运**最终结果**（往往远小于输入）。代价从\"搬输入\"变成\"搬输出\"。\n",
"\n",
"若两表没有按连接键分区，就必须先 **shuffle**（重分布）：按连接键的哈希值把每张表的记录\n",
"发到对应节点。按本模型估算，两张表都要搬：\n",
"\n",
" persons: 5 000 × 60 B = 300 000 B ≈ 293 KB\n",
" books : 10 000 × 52 B = 520 000 B ≈ 508 KB\n",
" 合计 ≈ 801 KB\n",
"\n",
"这比半连接归约（383 KB）还贵一倍以上。**这就是\"分区方式决定连接代价\"的直接体现。**\n",
"\n",
"代价还包括：分区本身是一次全量重排（离线成本），以及数据倾斜时某些节点会明显更慢。\n",
"\n",
"答案要点
\n",
"\n",
"成立，但要**选对方向**。半连接归约的原则是：**先算选择率，谁过滤后更小，就从谁出发投影。**\n",
"\n",
"过滤条件在 books 一侧时的做法：\n",
"\n",
"1. s2 本地过滤 books（`WHERE price < 20`），取出 `DISTINCT author`；\n",
"2. 把 author 投影发给 s1；\n",
"3. s1 用它半连接过滤 persons，只保留\"写过便宜书\"的人；\n",
"4. 把过滤后的 persons 发回 s2（或留在 s1，看哪边后续更划算）；\n",
"5. 在某一侧完成最终连接。\n",
"\n",
"一般化：**两侧都有过滤条件时，两侧都先本地过滤再投影**，这样两端的投影都是最小的。\n",
"\n",
"判断方向时不能只看行数，还要看**投影的字节数**：一侧行数少但连接键很宽，另一侧行数多但键很窄，\n",
"后者可能更划算。\n",
"\n",
"答案要点
\n",
"\n",
"它是**允许被广播的表的大小上限**，默认 10485760 字节 = 10 MB。\n",
"\n",
"- 优化器读表的统计信息估计大小：**估计值 ≤ 10 MB** → 走 broadcast join（把小表发到每个 executor，各 executor 本地连接）；\n",
"- **超过阈值** → 改用 Sort-Merge Join，两侧都按连接键 shuffle 重分布，再排序归并连接；\n",
"- 设为 **−1** 则彻底关闭广播。\n",
"\n",
"补充：自 **Spark 3.2.0** 起 **AQE**（Adaptive Query Execution）默认开启，\n",
"它能在运行时拿到真实的 shuffle 统计后，把原本计划的 Shuffle Join **动态转成** Broadcast Join。\n",
"这是\"估计不准就补救\"的机制，也说明静态阈值只是第一道防线。\n",
"\n",
"答案要点
\n",
"\n",
"**能测出**：每种策略**搬了多少数据**（字节数 / 行数），以及不同策略的结果集是否一致。\n",
"这两项正是本实验要建立的直觉——\"优化目标从算得快变成少搬数据\"。\n",
"\n",
"**测不出**：\n",
"\n",
"- 传输**耗时**与带宽竞争：没有序列化、没有网络栈、没有排队，`send` 是零成本的；\n",
"- 节点间的**并行加速**：这里是同一进程里顺序执行，广播换来的并行度体现不出来，\n",
" 所以\"广播虽然多传 2 份但整体更快\"这个结论在本 notebook 里看不到；\n",
"- 内存压力：真实系统里广播的小表要放进每个 executor 的内存，装不下会 OOM\n",
" （这也是为什么要有 10 MB 阈值）；\n",
"- **数据倾斜**：真实 shuffle 里某个热点键会让一个节点慢很多，这里没有。\n",
"\n",
"要看这些，得上真正的分布式引擎（Spark / Trino），用 `EXPLAIN` 观察优化器实际选了哪种分布方式。\n",
"\n",
"