# python笔试题

**Repository Path**: zhao-huanting/python-coding-test

## Basic Information

- **Project Name**: python笔试题
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-08-07
- **Last Updated**: 2026-08-07

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Python 后端笔试题（数据库方向）— 文献检索系统

> 数据存在 **Elasticsearch**。本仓库**不需要连真实 ES**：内置了一个内存版 mock（`mock_es.py`），
> 核心代码基于标准库即可运行；生产环境把 `FakeElasticsearch` 换成官方 `Elasticsearch` 客户端即可，调用代码无需改动。

---

## 运行方式

```bash
# 核心代码零依赖，直接跑演示（覆盖一、二、三部分 + 迁移）
python demo.py

# 跑测试（需要 pytest）
python -m pytest          # 15 个用例全绿

# 连接真实 ES（可选）：pip install elasticsearch 后
#   from elasticsearch import Elasticsearch
#   es = Elasticsearch("http://localhost:9200")
#   再把 demo / tests 里的 FakeElasticsearch() 换成 es 即可，queries/loader 不用改。
```

要求 Python 3.8+（依赖 `dict[str, ...]` 注解风格）。

## 项目结构

```
.
├── mock_es.py          # 内存版 ES：search/bulk/reindex/alias/聚合，与 elasticsearch-py API 对齐
├── queries.py          # 【一】ES 查询封装：全文检索 / 按年聚合 / 多条件过滤
├── inverted_index.py   # 【二】倒排索引：add / search / search_and / delete(tombstone)
├── loader.py           # 【三】CSV → ES 批量加载（流式 + 分批 + 幂等）+【附加】mapping 迁移
├── demo.py             # 一键演示三部分
├── sample.csv          # 演示用的 CSV（含一行坏数据 bad_year）
├── tests/              # pytest 用例（15 个）
└── requirements.txt    # 依赖说明
```

---

## 一、ES 查询封装（`queries.py`）

文档结构（`INDEX = "papers"`）：

```python
{
    "pmid": 12345678, "pub_year": 2024,
    "title": "Article Title", "abstracts": "...",
    "authors": ["Alice", "Bob"], "pub_type": "Journal Article"
}
```

| 能力 | 函数 | 说明 |
|---|---|---|
| 1 关键词全文检索 | `full_text_search(es, keyword, page, page_size)` | `multi_match` 在 `title`+`abstracts` 上搜索，`from/size` 分页，`track_total_hits=True` 取精确总数，返回每条 `_score` |
| 2 按年聚合 | `yearly_stats(es, keyword, years=3)` | 近 3 年（含当年）逐年命中数，从新到旧；空年份补 0 |
| 3 多条件过滤 | `search_with_filters(es, keyword, pub_year, pub_type, ...)` | 关键词 + 年份 + 类型，三者 AND |

关键点：

- **filter 上下文**：年份 `range` 和类型 `term` 都放在 `bool.filter`，不进 `must`，因此不影响相关度评分（见下方简答）。
- 按年聚合用 `terms` 聚合对 `pub_year` 分组计数，外层 `range` 过滤把窗口限制在近 3 年；`size: 0` 不取命中明细，只拿聚合结果。

### 简答：为什么年份和文献类型走 filter 不走 query？

1. **语义不同**。query 上下文会参与**相关度评分**（参与计算 `_score` 并影响排序），而 filter 上下文是纯布尔过滤，只决定"包含/排除"，不评分。年份、文献类型是精确的元数据约束——用户搜 `"cancer"` 时，期望排序只由"标题/摘要与 cancer 的相关程度"决定，而不该因为"这篇是 2024 年"就抬高或压低分数。
2. **性能**。filter 上下文的结果会被 ES 缓存（filter 命中集按段缓存为 bitset），同一过滤条件重复查询（例如仪表盘反复刷某年数据）命中缓存，更快。
3. **可维护性**。把"评分条件"和"过滤条件"分开，语义清晰：以后加/改过滤条件不会意外扰动排序。

---

## 二、倒排索引（`inverted_index.py`，纯标准库）

四个能力全部实现：

```python
ii = InvertedIndex()
ii.add(1, "Hello, World!")                 # 分词: hello, world
ii.add(2, "hello hello world python")
ii.search("hello")              # -> [1, 2]          升序
ii.search_and("hello", "python")  # -> [2]           交集
ii.delete(1)                    # tombstone 软删
ii.search("hello")              # -> [2]             跳过已删除
```

内部结构：

```python
_postings:  dict[str, list[int]]   # term -> 升序 doc_id posting list（二分插入/删除保持有序）
_docs:      dict[int, set[str]]    # doc_id -> 该文档的 term 集合（add 覆盖时据此移除旧倒排记录）
_tombstones:set[int]               # 软删除标记
```

实现要点：

- **覆盖语义**：`add` 同一 `doc_id` 时，先按 `_docs` 里记录的旧 term 从各 posting list 中移除该 doc，再写入新分词，最后更新 `_docs`。
- **AND 交集**：`search_and` 用**双指针归并**两条有序 posting list，不遍历全部文档，也不对每篇文档做集合查找。
- **软删**：`delete` 只加 tombstone；`search`/`search_and` 在返回时跳过 tombstone 中的 doc。重新 `add` 会复活。

### 简答

**1. AND 交集 vs 遍历全部文档，复杂度差多少？**

- 交集：`O(|P1| + |P2|)`，`P1/P2` 是两个 term 的 posting list 长度，只遍历命中的候选。
- 遍历全部文档：`O(N)`，对全部 `N` 篇文档逐一判断是否同时包含两个 term。
- 差距：当 term 常见、posting list 接近 `N` 时差距小；当 term 相对冷门（posting list ≪ N，最常见的情况）时，交集快 1~2 个数量级，且不依赖"每文档是否命中"的额外数据结构。倒排索引的核心价值就在于此：**让查询代价只和命中集合大小相关，而不是和全集相关**。

**2. 三个核心方法的时间复杂度？**

| 方法 | 复杂度 | 说明 |
|---|---|---|
| `add` | `O(|text| + k·log|P|)` | 分词 `O(|text|)`；对 `k` 个不重复 term 在有序 posting list 二分插入/删除各 `O(log|P|)`（覆盖时旧 term 删除同样是二分） |
| `search` | `O(|P|)` | 遍历该 term 的 posting list 过滤 tombstone |
| `search_and` | `O(|P1| + |P2|)` | 双指针归并两条有序 posting list |

（若忽略 tombstone 遍历，`search` 可用二分 + 计数近似 `O(log|P|)`，但按题目要求返回完整列表，`O(|P|)` 是基准。）

**3. ES 的 segment merge 解决什么问题？tombstone 在 merge 时起什么作用？**

- **segment merge 解决的问题**：ES/Lucene 的段（segment）是底层不可变文件。写入越多段越多，查询要并段搜索（慢），文件句柄、内存、IO 开销都增大。merge 把多个小段合并成大段，从而减少段数、提升查询速度、回收磁盘空间。
- **tombstone 在 merge 中的作用**：Lucene 删除文档不立即物理删除，而是在 `.del` 文件里记录"墓碑"。merge 生成新段时，**被标记删除的文档直接不写入新段**，实现真正的物理删除，`.del` 墓碑也随之作废。所以 tombstone 既是"软删的快速路径"，也保证了 merge 时能最终回收空间。

---

## 三、CSV → ES 批量加载（`loader.py`）

接口：

```python
result = load_csv(es, "sample.csv", index=ALIAS, batch_size=500)
# LoadResult(success=成功条数, failures=[{"line": 行号/batch, "reason": 原因}, ...])
```

实现要点：

- **流式读**：`csv.DictReader` 逐行迭代，整个文件不会一次进内存。
- **批量写**：按 `batch_size`（默认 500）攒满 `operations` 列表后调 `_bulk`，`1 条文档 = meta + source 两行`，所以满 `batch_size * 2` 条 flush 一次；**文件读完把不满一批的尾部也 flush**。
- **解析失败跳过**：`pub_year` 非整数、`pmid` 非整数都抛 `ValueError`，记 `{"line": 行号, "reason": 原因}` 后 `continue`，不影响其他行（表头占第 1 行，数据行号从 2 开始）。
- **authors 切分**：按 `;` 切数组并 strip 空白。
- **幂等**：文档 `_id = str(pmid)`，`bulk` 的 `index` action 是覆盖写，重跑同一文件结果一致。
- **失败明细**：解析失败记行号；bulk 响应里有 `error` 的 item 也记入（带 `batch` 定位 + `id` + 原因），成功条数按响应逐条累计。

### 【附加】mapping 迁移（`migrate_pub_type_to_keyword`）

```python
n = migrate_pub_type_to_keyword(es, "papers_v1", "papers_v2", alias="papers")
```

三步：

1. **建新索引**：`papers_v2` 的 mapping 里 `pub_type` 为 `keyword`（精确匹配/聚合）；
2. **reindex**：`{"source": {"index": "papers_v1"}, "dest": {"index": "papers_v2"}}` 全量搬运，返回搬运文档数；
3. **原子切换 alias**：一次 `update_aliases` 同时 `remove(v1) + add(v2)`，alias 从旧索引切到新索引。全程上层通过 alias 读写，**无感知**；旧索引留作回滚，确认后可删。

### 简答：bulk 部分失败时怎么处理？

`_bulk` 不是全有全无，**响应里逐条给出每个 item 的状态**，所以"部分失败"的处理就是**逐条甄别、分类重试**：

1. **解析响应**：`resp["errors"]` 为 true 时，遍历 `resp["items"]`，收集带 `error` 的 item（拿到 `index/id/error.type/error.reason`）。成功的照常计数。
2. **分类失败**：
   - **临时失败**（429 限流、`es_rejected_execution_exception` 熔断、连接/超时）：指数退避重试（如 1s/2s/4s，最多 3 次），仍失败进**死信/错误表**，由后续定时任务补偿。
   - **永久失败**（mapping 冲突 400、文档格式错、版本冲突）：重试无意义，直接记录 `id + 原因` 进失败明细，跳过继续处理后续批次，最后汇总报告。
3. **落地**：本仓库 `loader._flush_bulk` 即把失败 item（含 `batch` 定位、`id`、`reason`）收集进 `LoadResult.failures` 返回；生产上可选用官方 `elasticsearch.helpers.bulk(raise_on_error=False)`，用 `on_error` 钩子做同样的事。

---

## 四、CSV → ES 同步方案设计（只写方案）

> 场景：每天有新 CSV 落盘，loader 增量同步到 ES。首日全量 100 万条，每日增量 1~5 万条；
> 关键字段 `pmid`（唯一），另有单列删除清单 `deleted.csv`。

### 4.1 增量识别

每日增量由两部分组成，分别处理：

| 输入 | 处理 | ES API | 说明 |
|---|---|---|---|
| 新增/修改的 CSV | 逐条构造 `{"index": {"_index": ..., "_id": pmid}} + source`，按批 `_bulk` | `bulk`（`index` action） | `index` 是 upsert：存在则覆盖，不存在则新增，天然幂等 |
| `deleted.csv` | 每行一个 pmid，构造 `{"delete": {"_index": ..., "_id": pmid}}` 批量删除 | `bulk`（`delete` action） | 删除不存在的 id 返回 `not_found`，不算错误，仍幂等 |

> 为什么用 `bulk` 的 `delete` 而不是 `delete_by_query`？`bulk delete` 逐条可重试、可记录失败项、可控批量大小；`delete_by_query` 一把梭，中断/部分失败难恢复。

**为什么 `pmid` 当 `_id` 而不是自增 ID？**

1. **幂等/去重**：同一文献重复同步不会产生重复文档——重跑同一文件，`_id` 相同，`index` 覆盖写，最终状态一致；自增 ID 每次同步都新建，必然重复。
2. **天然 upsert 主键**：新增/修改/删除都按同一个 `pmid` 定位，无需先查询判断文档是否存在。
3. **业务主键即文档主键**：`pmid` 全局唯一，直接作为 ES 主键，语义清晰、便于跨系统对齐。

### 4.2 写入性能（首次 100 万条怎么调最快）

至少这几项（按收益排序）：

1. **`index.number_of_replicas = 0`**：全量期间关闭副本，避免每写一份就复制一份的 IO/网络开销；**灌完恢复默认（1）**。
2. **`index.refresh_interval = -1`（或 `30s`）**：灌入期间关闭/拉大 refresh，攒着分段批量刷盘；**灌完恢复 `1s`**。
3. **调大 bulk size**：单批压到 **1~8 MB** 或 **1000~5000 条/批**（按文档大小实测），减少往返。
4. **客户端并发**：多线程/多进程并发发 bulk（如 8~16 并发），控制并发数并监听 429 做退避重试。
5. **translog 策略**：调大 `index.translog.sync_interval` / `index.translog.flush_threshold_size`；可接受极端故障丢少量数据时设 `index.translog.durability=async`，减少每次刷盘。
6. **灌完收尾**：恢复 `refresh_interval=1s`、`replicas=1`，执行一次 `POST _forcemerge`（减少段数），验证文档数达标。

伪代码（使用官方 helpers）：

```python
from elasticsearch import Elasticsearch, helpers

es = Elasticsearch("...")
es.indices.put_settings(index=idx, body={"index": {"refresh_interval": -1, "number_of_replicas": 0}})
success, _ = helpers.bulk(es, gen_actions(csv_stream, idx), chunk_size=2000,
                          request_timeout=120, raise_on_error=False)
es.indices.put_settings(index=idx, body={"index": {"refresh_interval": "1s", "number_of_replicas": 1}})
es.indices.forcemerge(index=idx, max_num_segments=1)
```

### 4.3 断点恢复

同步跑到一半进程挂了，重启要**续传**，需要持久化"进度状态"：

- **持久化什么**：已处理的文件 + 每个文件已成功提交到的位置（行号/已写 pmid），以及 `deleted.csv` 的进度。**关键是 checkpoint 只在"整批 bulk 成功提交后"才推进**，否则会漏数据。
- **存在哪里**：选项——本地磁盘文件（如 `checkpoint.json`）、数据库表、Redis，或 ES 自身一个 `sync_checkpoint` 索引。推荐 **本地文件 + 定期落盘**（简单、无外部依赖），多机部署时换数据库表/Redis。
- **checkpoint 结构例子**：

```json
{
  "source_dir": "inbox/",
  "files": {
    "2026-08-06.csv": {
      "status": "processing",          // processing | done
      "last_row": 230000,              // 已成功提交到的行号（含）
      "last_pmid": "12345999",         // 便于快速校验
      "commit_point": 204800           // 最后一次 bulk 提交的字节/行位置
    },
    "deleted.csv": { "status": "done", "last_row": 10000 }
  },
  "updated_at": "2026-08-06T19:00:00Z"
}
```

- **续传流程**：启动 → 读 checkpoint → `status=processing` 的文件从 `last_row + 1` 继续读（CSV 按行读，行内不跨行拆分，保证不重不漏）→ `status=done` 的文件跳过 → 每成功提交一批即更新并落盘 checkpoint。

### 4.4 幂等

**同一文件处理两次，ES 最终状态完全一致**，靠四条：

1. **`_id = pmid`**：bulk `index` 是覆盖写，新增/修改重跑 = 覆盖，不产生重复文档。
2. **`delete` 幂等**：删除不存在的 `_id` 返回 `not_found`，在 bulk 里不算错误，重复删结果一致。
3. **固定处理顺序**：同一文件总是"先新增/修改，后删除"（或按文件顺序），保证最终态收敛到同一结果。
4. **checkpoint 兜底**：断点续传 + 整批提交才推进 checkpoint，即使某批重复执行，也因 `_id` 幂等而无副作用。

如需更严格，可为每个文件记录内容 `sha256`，发现同一 hash 直接跳过——防的是"文件改名换壳但内容相同"的场景。

---

## 附：Mock 说明

`mock_es.FakeElasticsearch` 是一个内存版 ES，与 `elasticsearch-py` 的 API 对齐，只实现了本作业用到的子集：

- `search`：`match / multi_match / term / terms / range / bool(must/filter/must_not/should)`、`from/size`、`track_total_hits`、`sort`、`aggs.terms`；
- 写入：`index / delete / bulk`；
- 索引/别名：`indices.create/exists/delete/refresh`、`put_alias/delete_alias/update_aliases`、`reindex`。

**限制**：相关度分数是"简化版 BM25"（tf·idf，逐字段 idf），与真实 ES 的 `english` analyzer + BM25 数值上有差异；聚合只实现了 `terms`。这些只影响数值演示，不影响查询 DSL 的正确性与封装层（`queries.py`/`loader.py`）本身。