切、算、并。
509 页,九份幻灯片;一格一页,滑过去看章名。
导论 1–9 · chap1 10–72 · chap2 73–140 · chap3 141–162 · chap4a 163–190 · chap4b 191–296 · chap5a 297–364 · chap5b 365–414 · chap6a 415–509
Plate I
复制机
chap3 · p.141–162
大不是体积,是失衡
chap1 · p.10–72
数据长得比摩尔定律快,机器只能靠便宜堆数量。当一千台机器平均每天都坏一台,「不会坏」就不再是设计前提。
讲义把这件事拆成四个 V:Volume、Variety、Velocity、Veracity。量只是第一个。
上海五万辆计程车,每 15 秒回报一次位置,一天就是 2.88 亿个点。这些点不是为了被储存而存在,是为了被算。
把文件切开,撒在会坏的机器上
chap3 · p.141–162
128MB 一块,默认三份。第一份靠近写入端,第二份跨到别的机架,第三份回到第二个机架的另一台。NameNode 只记位置,DataNode 才真的存数据。
切块器
拖文件大小,看它变成几个区块、几份副本。
三份副本落在两个机架:第一份在写入端,第二、三份跨到另一个机架的另一台。
没有脚本也能读:1 TB 的文件 = 8,192 个 128MB 区块 × 3 份副本 = 24,576 份复本;单一台机器或单一机架损坏,数据仍然可读。
这不是备份,是概率。一台机器三年下线一次,一千台就是每天一台;副本让「坏掉」变成例行工作,而不是事故。
Hadoop 的原则是反过来的:不要叫机器变得更可靠,叫软体接受它会坏。所以它把程序搬到数据旁边,不是把数据搬到程序旁边。HDFS 管储存、YARN 管资源;两层分开,储存与运算才能各自长大。
同一个键,永远去同一个 reducer
chap2 · p.73–140
map 逐行产生 (键, 值);partitioner 用键的哈希决定它去哪;同一个键的所有值在 reducer 碰头、排序、合并。
洗牌机器
点任何一个词,看它被送到哪一个 reducer。切换 reducer 数量,分布会变。
没有脚本也能读:四个文件里 Alice 出现 4 次、Bob 5 次、Carol 3 次;同一个词的全部出现次数一定被送到同一个 reducer。
副作用很有用:每个 reducer 收到的键是排好序的,所以 sort 只要一个 range partitioner,join 只要把两边的数据打上 LEFT / RIGHT 标签送进同一个 reducer。后来的 Hive 与 Pig 把 SQL 叠回 MapReduce,HadoopDB 则把两者缝在一起。
RDBMS 假设查询会跑完,MapReduce 假设它一定会掉。Google 自己的数字是每个 job 平均失败 1.2 次,所以它把「重跑」写进模型里,而不是写在维运手册里。
| 差异 | 并行 RDBMS | MapReduce |
|---|---|---|
| schema | 必须有;换来逻辑独立与数据共享 | 不需要;结构由写程序的人自己管 |
| 表达方式 | SQL 加 UDF,宣告式 | 程序语言,自己排数据流 |
| 排程 | 编译期,有优化器 | 执行期,没有全域优化 |
| 数据流 | push,中间结果不落地 | pull,中间结果落地 |
| 粒度 | 查询级 | 储存区块级 |
少搬一次,就快一百倍
chap4a–b · p.163–296
RDD 不可变,所以坏掉的 partition 可以从 lineage 重算;narrow dependency 可以叠成同一个阶段,wide dependency 一定要洗牌。快不是因为它更聪明,是因为中间结果留在内存。
DAG 分阶段
按「执行」看 Spark 真的跑了什么;按「弄坏一个分区」看它怎么重算。
没有脚本也能读:map 与 filter 之间是 narrow dependency,可以合成一个阶段;reduceByKey 之前是 wide dependency,必须洗牌,也是阶段的边界。
没有 action,上面那条链什么都不做。lazy 不是省事,是让 Spark 有机会把 map 与 filter 排进同一次扫描;而 take(10) 算到第十个就停,后面的数据连读都不读。分区器决定一列去哪里:reduceByKey 用哈希,sortByKey 用范围。
上面那些 map、filter、reduce,语法不是 Spark 发明的,是 Scala 集合搬上集群。不可变的 val 与高阶函数,让「重算」变成安全的默认。
val lines = sc.textFile("hdfs://log.txt")
val errs = lines.filter(_.contains("Exception"))
val byHost = errs
.map(l => (hostOf(l), 1))
.reduceByKey(_ + _)
byHost.take(10) // action:到这里才真的算
先砍掉不可能,再数剩下的
chap5a–b · p.297–414
支持度是「出现几次」,置信度是「出现之后,真的会买的比率」。单调性让候选集一层层变小:Apriori 拿多次扫描换内存,FP-tree 拿一次扫描换一棵树。
两个算法,同一张表
五笔交易,支持度门槛 3;切到 FP-tree,看它怎样不扫第二次。
| TID | A | B | C | D | E |
|---|---|---|---|---|---|
| t1 | 1 | 0 | 0 | 1 | 0 |
| t2 | 1 | 1 | 0 | 1 | 1 |
| t3 | 0 | 1 | 1 | 0 | 0 |
| t4 | 1 | 1 | 1 | 1 | 1 |
| t5 | 0 | 1 | 1 | 0 | 1 |
没有脚本也能读:支持度 3 之下,频繁 2-项集只有 {A,D}、{B,C}、{B,E};{B,C}→E 的支持度只有 2,所以它连候选都不是。
找所有频繁项集本身是 NP-hard,所以后面每个算法都在换:换扫描次数、换内存、或换一点正确性。PCY 用哈希桶先滤掉不可能的配对;SON 把数据切块各跑一次,再用单调性取并集;Toivonen 用负边界赌一把,赌输就换一组样本重来。
维度一多,邻居就消失
chap6a · p.415–509
在很高的维度里,任两点的距离几乎一样长。与其先降维再聚类,不如直接问:哪几个维度的组合,本身就有密集的格子?
子空间,示意
切换看哪些维度。两维一起看时,族群会消失。
密集单位 0 / 36。每格密度最高 4.2%,离 20% 很远。
KL 转换
讲义的四个点。按「转轴」,看它们投影到哪里。
按「转轴」,把坐标系转到特征向量的方向。
没有脚本也能读:密集单位是落在同一格超过 20% 的数据;两维一起看时没有密集单位,只看 Age 时 A3 与 A5 各 25%。KL 部分:四个点投影到 λ=1 的方向得 0、0、−1.41、1.41。
CLIQUE 用 apriori 找密集单位,再把相连的格子并成集群;熵方法把「好聚类」写成 H(X) < ω(八匹马等概率是 3 bit,确定是 0 bit);KL 转换把坐标轴转到方差的方向,最小特征值那条,常常就是答案。
整条路的帐单
本页的计算
把一份 1 TB 的数据推过三个闸:写入、洗牌、留在内存。帐单只记一件事:总共搬动了多少 bytes。
到目前为止,共搬动
0 B
0× 输入
写入磁盘0 B
跨网络0 B
再写一次(没有 cache)0 B
没有脚本也能读:1 TB 输入,三份副本 = 写 3 TB、跨网络 2 TB;洗牌量取决于 combiner;没有 cache 时中间结果要再落地一次。