数据库（五）：NoSQL——文档型、键值型、列式与图数据库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// MongoDB 中的一个用户文档
{
  "_id": ObjectId("507f1f77bcf86cd799439011"),
  "email": "alice@example.com",
  "name": "Alice Chen",
  "addresses": [
    {
      "type": "home",
      "street": "123 Main St",
      "city": "San Francisco",
      "state": "CA",
      "zip": "94105"
    },
    {
      "type": "work",
      "street": "456 Market St",
      "city": "San Francisco",
      "state": "CA",
      "zip": "94103"
    }
  ],
  "preferences": {
    "newsletter": true,
    "theme": "dark",
    "language": "en"
  },
  "created_at": ISODate("2023-11-15T10:30:00Z")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// 连接 MongoDB
const db = client.db("ecommerce");
const users = db.collection("users");

// 创建（Create）
await users.insertOne({
  email: "alice@example.com",
  name: "Alice Chen",
  addresses: [{ type: "home", city: "San Francisco" }],
  created_at: new Date()
});

// 读取（Read）
const user = await users.findOne({ email: "alice@example.com" });

// 投影读取（仅返回指定字段）
const userBasic = await users.findOne(
  { email: "alice@example.com" },
  { projection: { name: 1, email: 1, _id: 0 } }
);

// 更新（Update）：添加新地址
await users.updateOne(
  { email: "alice@example.com" },
  { $push: { addresses: { type: "work", city: "Oakland" } } }
);

// 更新（Update）：递增计数器
await users.updateOne(
  { _id: userId },
  { $inc: { login_count: 1 }, $set: { last_login: new Date() } }
);

// 删除（Delete）
await users.deleteOne({ email: "alice@example.com" });

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// 查找位于旧金山且偏好深色主题的用户
const result = await users.find({
  "addresses.city": "San Francisco",
  "preferences.theme": "dark"
}).sort({ created_at: -1 }).limit(10).toArray();

// 查找最近 30 天内创建的用户
const recent = await users.find({
  created_at: { $gte: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000) }
}).toArray();

// 全文搜索（需先建立 text 索引）
await users.createIndex({ name: "text", email: "text" });
const searchResults = await users.find({
  $text: { $search: "alice chen" }
}).toArray();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

// 计算上季度各品类商品营收
const pipeline = [
  // 阶段 1：筛选上季度已完成订单
  { $match: {
    status: "completed",
    created_at: { $gte: new Date("2023-10-01") }
  }},
  // 阶段 2：展开 order items 数组（每项生成一个文档）
  { $unwind: "$items" },
  // 阶段 3：关联产品详情
  { $lookup: {
    from: "products",
    localField: "items.product_id",
    foreignField: "_id",
    as: "product"
  }},
  // 阶段 4：展开 product 数组
  { $unwind: "$product" },
  // 阶段 5：按品类分组
  { $group: {
    _id: "$product.category",
    total_revenue: { $sum: { $multiply: ["$items.quantity", "$items.price"] } },
    order_count: { $sum: 1 },
    avg_order_value: { $avg: { $multiply: ["$items.quantity", "$items.price"] } }
  }},
  // 阶段 6：按营收降序排序
  { $sort: { total_revenue: -1 } },
  // 阶段 7：重命名输出字段
  { $project: {
    category: "$_id",
    total_revenue: { $round: ["$total_revenue", 2] },
    order_count: 1,
    avg_order_value: { $round: ["$avg_order_value", 2] },
    _id: 0
  }}
];

const results = await orders.aggregate(pipeline).toArray();
// [
//   { category: "Electronics", total_revenue: 45230.50, order_count: 312, avg_order_value: 145.00 },
//   { category: "Books",       total_revenue: 12890.00, order_count: 567, avg_order_value: 22.73 },
//   ...
// ]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# 字符串（Strings）：最基础的键值对
SET user:1:name "Alice Chen"
GET user:1:name                    # "Alice Chen"

SET page:home:views 0
INCR page:home:views               # 1（原子递增）
INCR page:home:views               # 2
INCRBY page:home:views 100         # 102

# 带过期时间（TTL）的字符串
SET session:abc123 '{"user_id":1,"role":"admin"}' EX 3600  # 1 小时后过期
TTL session:abc123                 # 3597（剩余秒数）

# 哈希（Hashes）：类似微型文档
HSET user:1 name "Alice" email "alice@example.com" login_count 42
HGET user:1 name                   # "Alice"
HGETALL user:1                     # name "Alice" email "alice@example.com" login_count "42"
HINCRBY user:1 login_count 1       # 43

# 列表（Lists）：有序集合（底层为链表）
LPUSH notifications:user:1 "New order #1234"
LPUSH notifications:user:1 "Payment received"
LRANGE notifications:user:1 0 9    # 最近 10 条通知
LLEN notifications:user:1          # 通知总数

# 集合（Sets）：无序唯一集合
SADD product:42:tags "electronics" "wireless" "bluetooth"
SMEMBERS product:42:tags           # {"electronics", "wireless", "bluetooth"}
SISMEMBER product:42:tags "wireless"  # 1（true）
# 集合运算
SINTER product:42:tags product:99:tags  # 两产品共有的标签

# 有序集合（Sorted Sets）：按 score 排序（极适合排行榜、排名）
ZADD leaderboard 1500 "player:alice"
ZADD leaderboard 2300 "player:bob"
ZADD leaderboard 1800 "player:carol"
ZREVRANGE leaderboard 0 2 WITHSCORES
# 1) "player:bob"    2) "2300"
# 3) "player:carol"  4) "1800"
# 5) "player:alice"  6) "1500"
ZRANK leaderboard "player:carol"   # 1（0-indexed，升序）
ZREVRANK leaderboard "player:carol" # 1（0-indexed，降序）

1
2
3
4
5
6
7

# redis.conf：启用双持久化以获得最高安全性
save 900 1        # 900 秒内 ≥1 个 key 变更则快照
save 300 10       # 300 秒内 ≥10 个 key 变更则快照
save 60 10000     # 60 秒内 ≥10000 个 key 变更则快照

appendonly yes
appendfsync everysec   # 每秒 fsync（良好平衡点）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 限流（滑动窗口）
# 每用户每分钟最多 100 次请求
MULTI
ZADD ratelimit:user:1 1702345678.123 "req-uuid-1"
ZREMRANGEBYSCORE ratelimit:user:1 0 1702345618.123  # 清除 >60s 的旧记录
ZCARD ratelimit:user:1  # 统计剩余请求数
EXPIRE ratelimit:user:1 60
EXEC

# 分布式锁（简化版）
SET lock:process-payments "" NX EX 30  # 获取锁，30 秒超时
# NX = 仅当 key 不存在时才设置
# 返回 OK 表示成功获取，nil 表示已被锁定

# 发布/订阅（Pub/Sub）
SUBSCRIBE channel:orders
PUBLISH channel:orders '{"order_id": 1234, "action": "created"}'

# 缓存旁路（Cache-aside）模式
# 1. 检查缓存
GET product:42
# 2. 若未命中，则查库并写入缓存
SET product:42 '{"name":"Widget","price":9.99}' EX 300  # 5 分钟 TTL

1
2
3
4
5
6
7
8
9

-- CQL（Cassandra Query Language）
CREATE TABLE user_activity (
    user_id     UUID,
    activity_date DATE,
    activity_time TIMESTAMP,
    activity_type TEXT,
    details       MAP<TEXT, TEXT>,
    PRIMARY KEY ((user_id), activity_date, activity_time)
) WITH CLUSTERING ORDER BY (activity_date DESC, activity_time DESC);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

-- 插入数据
INSERT INTO user_activity (user_id, activity_date, activity_time, activity_type, details)
VALUES (
    550e8400-e29b-41d4-a716-446655440000,
    '2023-12-15',
    '2023-12-15T14:30:00Z',
    'purchase',
    {'product_id': '42', 'amount': '149.99'}
);

-- 查询：获取某用户近期活动（高效——单分区查询）
SELECT * FROM user_activity
WHERE user_id = 550e8400-e29b-41d4-a716-446655440000
  AND activity_date >= '2023-12-01'
LIMIT 20;

-- 跨分区查询：应避免！（全集群扫描）
-- SELECT * FROM user_activity WHERE activity_type = 'purchase';
-- 此操作需 `ALLOW FILTERING`，将扫描整个集群

1
2
3
4
5

(Alice:Person {name: "Alice", age: 30})
    -[:FRIENDS_WITH {since: 2020}]->
(Bob:Person {name: "Bob", age: 28})
    -[:WORKS_AT {role: "Engineer"}]->
(Acme:Company {name: "Acme Corp", industry: "Tech"})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- SQL 中的 3 跳好友查询（痛苦）
SELECT DISTINCT p4.name
FROM friendships f1
JOIN friendships f2 ON f1.friend_id = f2.person_id
JOIN friendships f3 ON f2.friend_id = f3.person_id
JOIN people p4 ON f3.friend_id = p4.person_id
WHERE f1.person_id = 1
  AND p4.person_id != 1;
-- 每增加一跳，性能呈指数级下降
-- 在百万级用户的社交图中，此查询几乎不可行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

-- 创建超表（按时间自动分区）
CREATE TABLE metrics (
    time TIMESTAMPTZ NOT NULL,
    host VARCHAR(50),
    cpu_usage FLOAT,
    memory_usage FLOAT,
    disk_io FLOAT
);

SELECT create_hypertable('metrics', 'time');

-- 时间桶聚合（TimescaleDB 特有）
SELECT
    time_bucket('5 minutes', time) AS bucket,
    host,
    AVG(cpu_usage) AS avg_cpu,
    MAX(cpu_usage) AS peak_cpu
FROM metrics
WHERE time > NOW() - INTERVAL '1 hour'
GROUP BY bucket, host
ORDER BY bucket DESC;

-- 连续聚合（物化，自动刷新）
CREATE MATERIALIZED VIEW hourly_metrics
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS hour,
    host,
    AVG(cpu_usage) AS avg_cpu,
    percentile_cont(0.95) WITHIN GROUP (ORDER BY cpu_usage) AS p95_cpu
FROM metrics
GROUP BY hour, host;

-- 数据保留策略（自动删除 90 天前的数据）
SELECT add_retention_policy('metrics', INTERVAL '90 days');

1
2
3
4
5
6

查询: "如何部署 Python 应用？"
                                          余弦相似度
嵌入 → [0.23, -0.14, 0.87, ...]  ──────────────→  Top-K 结果
                                        0.94: "Flask 生产部署指南"
                                        0.91: "Python 应用 Docker 教程"
                                        0.87: "CI/CD 流水线配置"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

connections.connect(host="localhost", port="19530")

fields = [
    FieldSchema("id", DataType.INT64, is_primary=True),
    FieldSchema("text", DataType.VARCHAR, max_length=1000),
    FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=1536),
]
schema = CollectionSchema(fields)
collection = Collection("documents", schema)

# 创建 HNSW 索引
collection.create_index("embedding", {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 200},
})

# 搜索
results = collection.search(
    data=[query_embedding],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 100}},
    limit=10,
    output_fields=["text"],
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

CREATE EXTENSION vector;

CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1536)
);

-- HNSW 索引
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

-- 语义搜索
SELECT id, content, 1 - (embedding <=> $1) AS similarity
FROM documents
ORDER BY embedding <=> $1
LIMIT 10;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 文档模型
CREATE TABLE products (id SERIAL PRIMARY KEY, data JSONB);

-- 时序（TimescaleDB）
CREATE TABLE metrics (time TIMESTAMPTZ, value FLOAT);
SELECT create_hypertable('metrics', 'time');

-- 向量搜索（pgvector）
CREATE TABLE embeddings (id INT, vec vector(768));

-- 全文搜索（内置）
CREATE INDEX idx_fts ON articles USING GIN (to_tsvector('english', content));

1
2
3
4
5
6
7
8

// MongoDB：可调写的写关注（write concern）
await collection.insertOne(doc, {
  writeConcern: { w: "majority", j: true }  // CP 行为
});

await collection.insertOne(doc, {
  writeConcern: { w: 1 }  // AP 行为（仅主节点确认）
});

1
2
3
4
5
6
7

-- Cassandra：可调每条查询的一致性级别
-- Quorum 读 + Quorum 写 = 强一致性
SELECT * FROM users WHERE user_id = ? CONSISTENCY QUORUM;
INSERT INTO users (...) VALUES (...) USING CONSISTENCY QUORUM;

-- ONE = 快速但可能返回陈旧数据
SELECT * FROM users WHERE user_id = ? CONSISTENCY ONE;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- CockroachDB：语法像 PostgreSQL，扩展性如 Cassandra
CREATE TABLE orders (
    order_id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    user_id UUID NOT NULL,
    total DECIMAL(10,2),
    created_at TIMESTAMPTZ DEFAULT now()
);

-- 事务行为与 PostgreSQL 完全一致
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
UPDATE accounts SET balance = balance + 100 WHERE id = 2;
COMMIT;
-- 但数据通过 Raft 共识分布在多个节点上