slb: 分片负载均衡器

类似于 parallel --pipe --roundrobin，但是负载均衡是基于输入行哈希进行的。在子进程中执行键聚合时，这是至关重要的，因为只有其中一个分片包含给定的键。以下是一个在 16 个物理 CPU 机器上的单词计数示例

curl -o enwik9.bz2 https://cs.fit.edu/~mmahoney/compression/enwik9.bz2
bunzip2 enwik9.bz2
examples/clean.sh < enwik9 > enwik9.clean ; rm enwik9

/usr/bin/time -f "%e sec" awk -f examples/wc.awk enwik9.clean > wikawk.txt
# 203.97 sec

/usr/bin/time -f "%e sec" slb \
  --mapper 'tr " " "\n" | rg -v "^$"' \
  --folder "awk '{a[\$0]++}END{for(k in a)print k,a[k]}'" \
  --infile enwik9.clean \
  --outprefix wikslb.
# 6.20 sec

diff <(sort wikawk.txt) <(cat wikslb.* | sort) ; echo $?
# 0

这展示了在典型的 "map-reduce" 模式上的 "flatmap-fold" 模式。

每一行

a    b c d -> flatmapper 1
f g   a b -> flatmapper 2

被交给一个独立的平坦映射器 tr " " "\n" | rg -v "^$"，它在每一行上放一个单词

flatmapper 1 ->
a
b
c
d

flatmapper 2 ->
f
g
a
b

输出结果将逐行检查。每行的第一个单词会被哈希（在这种情况下，整行）。假设 hash(a) == hash(b) == 1 和 hash(c) == hash(d) == hash(g) == hash(f) == 0，我们将从每个平坦映射器中输入相应的密钥到一个或两个 awk '{a[$0]++}END{for(k in a)print k,a[k]}' 文件夹中。然后，输出将被写入输出文件。

a b a b -> awk 1 -> {a: 2, b: 2} -> outprefix1
f g c d -> awk 0 -> {f: 1, g: 1, c: 1, d: 1} -> outprefix0

特征频率计算

以下是一个计算大数据集稀疏 SVMlight 格式特征频率的示例，在大 KDD12 数据集上进行基准测试（假设已安装 ripgrep，GNU Parallel）。

echo 'will cite' | parallel --citation 1>/dev/null 2>/dev/null
curl -o kdd12.tr.bz2 "https://www.csie.ntu.edu.tw/~cjlin/libsvmtools/datasets/binary/kdd12.tr.bz2"
bunzip2 kdd12.tr.bz2
du -hs kdd12.tr 
# 17G     kdd12.tr
parallel --pipepart -a kdd12.tr wc -l | awk '{a+=$0}END{print a}'
# num rows: 119705032
parallel --pipepart -a kdd12.tr wc -w | awk '{a+=$0}END{print a}'
# num nnz: 1436460384 - 119705032 = 1316755352

/usr/bin/time -f "%e sec %M KB" awk -f examples/svm-featurecount.awk kdd12.tr > results-awk.txt
# 1032.18 sec 13721032 KB

/usr/bin/time -f "%e sec %M KB" slb \
  --mapper 'sed -E "s/^[^ ]+ //" | sed -E "s/:[^ ]+//g" | tr " " "\n" | rg -v "^$"' \
  --folder "awk '{a[\$0]++}END{for(k in a)print k,a[k]}'" \
  --infile kdd12.tr \
  --outprefix results-slb.
# 122.50 sec 881436 KB
# note above doesn't count child memory
# eyeballing htop, max memory use is ~12.3GB

# check we're correct
cat results-slb.* > results-slb && rm results-slb.*
sort --parallel=$(($(nproc) / 2)) -k2nr -k1n -o results-slb results-slb & \
sort --parallel=$(($(nproc) / 2)) -k2nr -k1n -o results-awk.txt results-awk.txt & \
wait

diff results-slb results-awk.txt >/dev/null ; echo $?
# 0

计算不同特征值

作为另一个类似的例子，我们可以计算每个特征的不同值的数量。特别是，对于每个特征，我们希望得到其不同值的总数与100的最小值（因为我们可能会倾向于将具有超过99个值的任何事物视为连续的）。

curl -o kdda.bz2 "https://www.csie.ntu.edu.tw/~cjlin/libsvmtools/datasets/binary/kdda.bz2"
bunzip2 kdda.bz2
du -hs kdda
# 2.5G    kdda

/usr/bin/time -f "%e sec %M KB" awk -f examples/svm-countdistinct.awk kdda > cdawk.txt
# 388.72 sec 23895104 KB

/usr/bin/time -f "%e sec %M KB" slb \
  --mapper 'sed -E "s/^[^ ]+ //" | tr " " "\n" | tr ":" " " | rg -v "^$"' \
  --folder "awk '{if(!(\$1 in a)||length(a[\$1])<100)a[\$1][\$2]=1}END{for(k in a)print k,length(a[k])}'" \
  --infile kdda \
  --outprefix cdslb.
# 26.79 sec 1499992 KB

diff \
  <(sort --parallel=$(($(nproc) / 2)) -k2nr -k1n cdawk.txt) \
  <(cat cdslb.* | sort --parallel=$(($(nproc) / 2)) -k2nr -k1n) \
  > /dev/null ; echo $?
# 0

安装

注意上述示例展示了工具的便利性

• 对于大数据集，并行化是必要的。
• 与等效的map-reduce相比，我们使用更少的内存、更少的时间和更少的代码。

最后一点成立，因为 slb 确保每个并行调用都接收一个 唯一的分区 的键空间。因此，我们使用更少的内存，因为每个文件夹只跟踪其自己的键空间的聚合，并且使用更少的代码，因为我们不需要编写合并两个映射的合并器。

要从 crates.io 本地安装，请运行

cargo install slb

开发相关

使用 ./test.sh 进行基本测试。

使用 cd slb && cargo publish 发布到 crates.io。