数据血缘

工作原理

血缘从三类来源采集：SQL 解析、编排器元数据和运行时事件。SQL 解析器提取输入输出表，并在能解析别名时提取列映射；编排器知道哪个作业在什么时间以什么参数运行；运行时集成发出带有输入输出数据集的开始和结束事件，形成运行级图。

图通常以节点（数据集、作业、列）和边（读取、写入、派生）存储。列级血缘更难，因为表达式、联合和用户自定义函数会遮蔽映射。许多系统在全局提供表级精度，只对简单转换提供列级精度。

用途

第一是调试：坏数字可以向上追溯到引入它的作业或源，并在修复发布前识别下游消费者。第二是影响分析：删除一列之前，找出读取它的每个报表和模型。第三是治理：证明个人数据只流经批准的系统，并在要求的地方被删除。

权衡

血缘需要在许多引擎中埋点，并且常常难以处理动态 SQL、临时表和运行时计算出的文件路径。存储和界面成本随图规模增长。过时的血缘比没有更糟，因为它制造虚假信心。列级精度昂贵，应优先用于受监管或高价值数据。

示例与误解

一个收入看板下降 3%。表级血缘显示看板依赖 fct_revenue，但无法指出哪个源列发生变化。列级血缘把下降追溯到一次货币转换连接：国家代码重命名后该连接丢失了行。没有这条边，团队会检查每一张上游表。

误解是血缘是一张画一次就够的图。真正的血缘基于事件、随管道版本化，并且只有反映产生该数据的那次运行时才有用。

代码示例

import java.util.List;
import java.util.Objects;

public class RecordClasses {

    public record Point(int x, int y) {
        public Point {
            if (x < 0 || y < 0) {
                throw new IllegalArgumentException(
                    "Coordinates must be non-negative");
            }
        }

        public double distanceTo(Point other) {
            return Math.sqrt(
                Math.pow(x - other.x, 2) +
                Math.pow(y - other.y, 2)
            );
        }
    }

    public record Rectangle(Point topLeft, Point bottomRight) {
        public int width() {
            return bottomRight.x() - topLeft.x();
        }
        public int height() {
            return bottomRight.y() - topLeft.y();
        }
        public int area() {
            return width() * height();
        }
    }

    public record Employee(
        String name,
        String department,
        double salary,
        List<String> skills
    ) {
        public Employee {
            Objects.requireNonNull(name);
            skills = List.copyOf(skills);
        }
    }

    public static void main(String[] args) {
        Point p1 = new Point(0, 0);
        Point p2 = new Point(3, 4);
        System.out.println("Distance: " + p1.distanceTo(p2));

        Rectangle rect = new Rectangle(p1, p2);
        System.out.println("Area: " + rect.area());

        Employee emp = new Employee("Alice", "Engineering", 95000,
            List.of("Java", "Python", "SQL"));
        System.out.println(emp);
    }
}

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())
}