1
# 用纯文件存储用户数据
2
def save_user(user_id, name, email):
3
    with open(f"users/{user_id}.txt", "w") as f:
4
        f.write(f"{name}\n{email}\n")
5

6
def read_user(user_id):
7
    with open(f"users/{user_id}.txt", "r") as f:
8
        lines = f.readlines()
9
        return {"name": lines[0].strip(), "email": lines[1].strip()}

1
-- 关系模型：用表表示实体和关系
2
CREATE TABLE users (
3
    id    BIGINT PRIMARY KEY,
4
    name  VARCHAR(100) NOT NULL,
5
    email VARCHAR(255) UNIQUE
6
);
7

8
CREATE TABLE orders (
9
    id         BIGINT PRIMARY KEY,
10
    user_id    BIGINT NOT NULL REFERENCES users(id),
11
    amount     DECIMAL(10,2),
12
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
13
);
14

15
-- 多对多关系通过中间表实现
16
SELECT u.name, COUNT(o.id) AS order_count
17
FROM users u
18
JOIN orders o ON u.id = o.user_id
19
GROUP BY u.id;

1
// 文档模型：一个订单包含所有相关信息
2
{
3
  "_id": "order_001",
4
  "user": {
5
    "id": "user_001",
6
    "name": "张三",
7
    "email": "zhangsan@example.com"
8
  },
9
  "items": [
10
    {"product": "数据库系统概念", "price": 89.00, "qty": 1},
11
    {"product": "DDIA", "price": 128.00, "qty": 1}
12
  ],
13
  "total": 217.00,
14
  "status": "shipped",
15
  "created_at": "2026-04-22T10:00:00Z"
16
}

1
// Neo4j Cypher：查询张三的朋友的朋友（二度关系）
2
MATCH (me:User {name: '张三'})-[:KNOWS]->(friend)-[:KNOWS]->(fof)
3
WHERE NOT (me)-[:KNOWS]->(fof) AND me <> fof
4
RETURN DISTINCT fof.name AS recommendation, COUNT(friend) AS mutual_count
5
ORDER BY mutual_count DESC
6
LIMIT 10

1
# Redis 键值操作
2
SET user:001:name "张三"
3
GET user:001:name
4
# → "张三"
5

6
# 哈希：模拟对象
7
HSET user:001 name "张三" email "zhangsan@example.com" age 28
8
HGET user:001 email
9
# → "zhangsan@example.com"
10

11
# 有序集合：排行榜
12
ZADD leaderboard 9500 "player_A"
13
ZADD leaderboard 8700 "player_B"
14
ZREVRANGE leaderboard 0 9 WITHSCORES

1
-- InfluxDB：写入时序数据
2
INSERT cpu,host=server01,region=us-west value=0.64 1422568543702900257
3

4
-- 查询最近 1 小时的平均 CPU 使用率
5
SELECT mean(value) FROM cpu
6
WHERE time > now() - 1h AND host = 'server01'
7
GROUP BY time(5m)

1
-- 声明式：只描述目标，不描述过程
2
SELECT name, age FROM users WHERE age > 30 AND city = '北京' ORDER BY age DESC;

1
# 命令式：精确描述执行过程
2
result = []
3
for user in users:
4
    if user.age > 30 and user.city == '北京':
5
        result.append(user)
6
result.sort(key=lambda u: u.age, reverse=True)

1
# MapReduce：统计各城市用户数
2
def mapper(user):
3
    yield (user.city, 1)
4

5
def reducer(city, counts):
6
    yield (city, sum(counts))
7

8
# 等价 SQL（简洁得多）
9
# SELECT city, COUNT(*) FROM users GROUP BY city;

1
// Cypher：模式匹配，直觉表达图遍历
2
MATCH (u:User {name: '张三'})-[:PURCHASED]->(p:Product)<-[:PURCHASED]-(other:User)
3
RETURN other.name, collect(p.name) AS shared_products
4
ORDER BY size(shared_products) DESC

1
# 行存储：读取整行，浪费 I/O
2
# 假设每行 100 字节，age 列占 4 字节
3
# 100 万行需要读取 100 MB，但只用到 4 MB
4

5
# 列存储：只读 age 列
6
# 100 万行只需读取 4 MB
7
# I/O 减少约 25 倍！

1
# Elasticsearch：全文检索 + 聚合分析
2
curl -X POST "localhost:9200/articles/_search" -H 'Content-Type: application/json' -d '
3
{
4
  "query": {
5
    "multi_match": {
6
      "query": "数据库性能优化",
7
      "fields": ["title^3", "content"]
8
    }
9
  },
10
  "aggs": {
11
    "by_category": {
12
      "terms": { "field": "category.keyword" }
13
    }
14
  }
15
}'

1
# 一个电商系统的多语言持久化架构
2
class OrderService:
3
    def __init__(self):
4
        self.mysql = MySQLClient()      # 核心业务数据（订单、用户、商品）
5
        self.redis = RedisClient()      # 缓存、会话、库存预扣
6
        self.es = ElasticsearchClient() # 商品搜索、日志分析
7
        self.mongo = MongoClient()      # 商品详情（Schema 灵活）
8
        self.influx = InfluxDBClient()  # 业务监控指标
9

10
    async def create_order(self, order):
11
        # 1. Redis 预扣库存
12
        await self.redis.decr(f"stock:{order.product_id}")
13
        # 2. MySQL 写入订单（事务保证）
14
        await self.mysql.insert("orders", order)
15
        # 3. ES 更新搜索索引
16
        await self.es.index("orders", order)
17
        # 4. InfluxDB 记录指标
18
        await self.influx.write("order_created", tags={"product": order.product_id})