← 中文资源

简体中文版本

数据建模

数据建模把业务规则和访问模式转化为实体、键、关系与约束,使存储能够拒绝非法状态,而不是把它们存下来。

阅读英文原文

工作原理

数据建模从系统必须回答的问题和绝不能违反的规则出发。在此基础上识别实体、选择标识符、确定哪些关系是强制的,并把不变量编码为主键、外键、唯一约束、检查约束或可空性。模型不只是一张图,它还定义归属、生命周期、保留和删除行为。

不同存储提供不同的原语。关系型引擎偏好表、连接和约束;文档存储偏好嵌入式聚合与应用层校验;宽列存储把分区键当作分布和查询的单位;图存储把边建模为一等记录。推理方式相似,但可用的约束能力差别很大。

从访问模式出发

在选择键之前,先列出主要的读写路径。一个写入分布均匀却无法回答主查询的分区键会迫使全表扫描;一个纸面上很优雅的范式模型,可能为了首页需要六次连接。记录基数、预期增长以及每条路径所需的一致性。

权衡

刚性模式能尽早拦截坏数据、简化报表,但每次变更都变成迁移。灵活模式能快速接纳新形状,却把校验推给服务层,并常常产生同一实体的多个版本。冗余提高读取速度,同时带来更新异常。没有免费的选项,只能选择自己能够运维的失败模式。

何时使用

任何比单个请求存活更久的状态都值得认真建模,尤其是多个团队共同读写的状态。对于短期缓存或原型,最小模型就足够,但要记录假设,以便日后重新审视。

误解与示例

常见误解是应用上线后模型就完成了。模型会随产品演进,迁移本身就是设计的一部分。例如订单在购买时保存收货地址快照:如果只引用客户当前地址,客户搬家就会静默改写历史。把地址复制到订单中保留快照,这是有意的反规范化,而不是意外。

代码示例

package main

import (
	"fmt"
	"math/bits"
)

type Bitset []uint64

func NewBitset(size int) Bitset {
	if size < 0 {
		panic("negative bitset size")
	}
	return make(Bitset, (size+63)/64)
}

func (b Bitset) Set(index int) {
	b[index/64] |= uint64(1) << (uint(index) % 64)
}

func (b Bitset) Clear(index int) {
	b[index/64] &^= uint64(1) << (uint(index) % 64)
}

func (b Bitset) Has(index int) bool {
	return b[index/64]&(uint64(1)<<(uint(index)%64)) != 0
}

func (b Bitset) Union(other Bitset) Bitset {
	size := len(b)
	if len(other) > size {
		size = len(other)
	}
	out := make(Bitset, size)
	for i := range out {
		var left, right uint64
		if i < len(b) {
			left = b[i]
		}
		if i < len(other) {
			right = other[i]
		}
		out[i] = left | right
	}
	return out
}

func (b Bitset) Intersect(other Bitset) Bitset {
	size := len(b)
	if len(other) < size {
		size = len(other)
	}
	out := make(Bitset, size)
	for i := range out {
		out[i] = b[i] & other[i]
	}
	return out
}

func (b Bitset) Count() int {
	total := 0
	for _, word := range b {
		total += bits.OnesCount64(word)
	}
	return total
}

func (b Bitset) Slice() []int {
	out := make([]int, 0, b.Count())
	for wordIndex, word := range b {
		for word != 0 {
			bit := bits.TrailingZeros64(word)
			out = append(out, wordIndex*64+bit)
			word &^= uint64(1) << uint(bit)
		}
	}
	return out
}

func main() {
	active := NewBitset(80)
	for _, id := range []int{1, 5, 64, 79} {
		active.Set(id)
	}
	blocked := NewBitset(80)
	blocked.Set(5)
	blocked.Set(9)

	fmt.Println("count:", active.Count())
	fmt.Println("has 64:", active.Has(64), "has 65:", active.Has(65))
	fmt.Println("members:", active.Slice())
	fmt.Println("union count:", active.Union(blocked).Count())
	fmt.Println("intersect:", active.Intersect(blocked).Slice())

	active.Clear(64)
	fmt.Println("after clear:", active.Count())
}
from collections.abc import Iterator
from typing import Generic, TypeVar

T = TypeVar("T")


class RingBuffer(Generic[T]):
    def __init__(self, capacity: int) -> None:
        if capacity < 1:
            raise ValueError("capacity must be positive")
        self._slots: list[T | None] = [None] * capacity
        self._head = 0
        self._count = 0

    @property
    def capacity(self) -> int:
        return len(self._slots)

    def __len__(self) -> int:
        return self._count

    def push(self, item: T) -> T | None:
        evicted: T | None = None
        if self._count == self.capacity:
            evicted = self._slots[self._head]
            self._head = (self._head + 1) % self.capacity
            index = (self._head + self._count - 1) % self.capacity
        else:
            index = (self._head + self._count) % self.capacity
        self._slots[index] = item
        if self._count < self.capacity:
            self._count += 1
        return evicted

    def items(self) -> Iterator[T]:
        for offset in range(self._count):
            value = self._slots[(self._head + offset) % self.capacity]
            if value is not None:
                yield value

    def latest(self, count: int) -> list[T]:
        if count < 0:
            raise ValueError("count must not be negative")
        if count == 0:
            return []
        return list(self.items())[-count:]


def main() -> None:
    buffer: RingBuffer[str] = RingBuffer(3)
    for event in ["a", "b", "c"]:
        print("push", event, "evicted", buffer.push(event))
    print("full:", list(buffer.items()))
    print("push d evicted:", buffer.push("d"))
    print("after:", list(buffer.items()))
    print("latest 2:", buffer.latest(2))
    print("length:", len(buffer))


if __name__ == "__main__":
    main()