← 中文资源

简体中文版本

星型模式

星型模式以一张存储可度量事件的事实表为中心,周围是反规范化的维度表,从而简化连接并加速分析聚合。

阅读英文原文

工作原理

星型模式把度量与上下文分开。事实表在声明好的粒度上每行代表一个事件,例如一条订单明细,并包含数量、收入等数值度量以及指向维度表的外键。维度表保存描述性属性:产品类别、客户地区、门店和日期。日期维度让分析人员无需解析时间戳就能按财季筛选。

形状像星,是因为每个维度都直接连接到事实表。反规范化维度,例如把类别和部门放在产品维度上,可以避免一连串连接。代价是描述性数据重复,但这些数据变化很少,在列式存储中压缩效果很好。

粒度就是契约

在添加度量之前先声明粒度。如果事实表混合了订单级和明细级的行,汇总收入就会重复计算。每个度量都必须在该粒度上有效,每个维度键都必须存在。没有明确粒度的事实表会变成一堆互不相关的数字,不同报表对它们有不同的解释。

何时使用

对销售、发货、会话、支付等稳定业务流程的报表和商业智能使用星型模式。它适合按维度筛选并扫描大量事实行的聚合查询。对于高度规范化的交易写入,保留源模式,在数据仓库或数据集市中构建星型模型。

权衡与示例

星型模式以存储和 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())
}