工作原理
星型模式把度量与上下文分开。事实表在声明好的粒度上每行代表一个事件,例如一条订单明细,并包含数量、收入等数值度量以及指向维度表的外键。维度表保存描述性属性:产品类别、客户地区、门店和日期。日期维度让分析人员无需解析时间戳就能按财季筛选。
形状像星,是因为每个维度都直接连接到事实表。反规范化维度,例如把类别和部门放在产品维度上,可以避免一连串连接。代价是描述性数据重复,但这些数据变化很少,在列式存储中压缩效果很好。
粒度就是契约
在添加度量之前先声明粒度。如果事实表混合了订单级和明细级的行,汇总收入就会重复计算。每个度量都必须在该粒度上有效,每个维度键都必须存在。没有明确粒度的事实表会变成一堆互不相关的数字,不同报表对它们有不同的解释。
何时使用
对销售、发货、会话、支付等稳定业务流程的报表和商业智能使用星型模式。它适合按维度筛选并扫描大量事实行的聚合查询。对于高度规范化的交易写入,保留源模式,在数据仓库或数据集市中构建星型模型。
权衡与示例
星型模式以存储和 ETL 工作量为代价换取读取简单。如果管道没有应用缓慢变化维度规则,维度可能发生漂移。雪花模式节省空间,但增加连接和复杂度;宽扁平表读取更快,却失去可复用的维度。
示例:fact_sales(order_line_id, date_key, product_key, store_key, quantity, revenue) 连接 dim_product 和 dim_date。按类别和季度查询收入只需两次连接,优化器还能裁剪日期分区。同样的数据若规范化到五张表,需要更多连接,也更容易出现错误的筛选条件。
代码示例
export class SegmentTree {
private readonly size: number;
private readonly tree: number[];
constructor(values: readonly number[]) {
if (!Array.isArray(values)) throw new TypeError("values must be an array");
for (const value of values) {
if (typeof value !== "number" || !Number.isFinite(value)) {
throw new TypeError("values must contain finite numbers");
}
}
this.size = values.length;
this.tree = new Array<number>(Math.max(4, this.size * 4)).fill(0);
if (this.size > 0) this.build(values, 1, 0, this.size - 1);
}
sum(left: number, right: number): number {
if (!Number.isInteger(left) || !Number.isInteger(right)) {
throw new TypeError("range bounds must be integers");
}
if (left < 0 || right >= this.size || left > right) throw new RangeError("invalid range");
return this.query(1, 0, this.size - 1, left, right);
}
update(index: number, value: number): void {
if (!Number.isInteger(index) || index < 0 || index >= this.size) {
throw new RangeError("index out of range");
}
if (typeof value !== "number" || !Number.isFinite(value)) {
throw new TypeError("value must be a finite number");
}
this.assign(1, 0, this.size - 1, index, value);
}
private build(values: readonly number[], node: number, start: number, end: number): void {
if (start === end) {
this.tree[node] = values[start] ?? 0;
return;
}
const middle = (start + end) >> 1;
this.build(values, node * 2, start, middle);
this.build(values, node * 2 + 1, middle + 1, end);
this.tree[node] = this.tree[node * 2] + this.tree[node * 2 + 1];
}
private query(node: number, start: number, end: number, left: number, right: number): number {
if (left <= start && end <= right) return this.tree[node];
const middle = (start + end) >> 1;
let total = 0;
if (left <= middle) total += this.query(node * 2, start, middle, left, right);
if (right > middle) total += this.query(node * 2 + 1, middle + 1, end, left, right);
return total;
}
private assign(node: number, start: number, end: number, index: number, value: number): void {
if (start === end) {
this.tree[node] = value;
return;
}
const middle = (start + end) >> 1;
if (index <= middle) this.assign(node * 2, start, middle, index, value);
else this.assign(node * 2 + 1, middle + 1, end, index, value);
this.tree[node] = this.tree[node * 2] + this.tree[node * 2 + 1];
}
}
const tree = new SegmentTree([2, 4, 6, 8, 10]);
console.log(tree.sum(1, 3));
tree.update(2, 0);
console.log(tree.sum(0, 4));
console.log(tree.sum(2, 2));
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())
}