更新为注意对聚合键的更改

说明

在匿名环境中，我有一个要求，即删除文件中出现次数少于指定次数的列中的所有值。将指定多个列，这些列要求值在该列中出现x次以上，并且空白是相对于一个列的。也就是说，如果x=4，值123在第1列中出现3次，在第4列中出现4次，那么第1列中的所有3次出现都必须为空，但在第4列中是允许的。
我已经使用了一种利用hadoop流媒体技术和perl的双通道方法解决了这个问题，这种方法在过去半年中一直很稳定。问题是提供了一个新文件，虽然我的进程将处理它，但在862容器（54节点）hadoop集群上处理它大约需要5天。我将解释我的解决方案/方法，并要求任何优化或替代方法，以适应这个问题，并允许优化运行时间。
正在处理的文件统计信息：
大小1.4tb
19427列
记录22.1英里
要素429336700000

二次进近（当前）

通过1

这个过程使用hadoop流，使用perl（v5.10.1）Map器和perl reducer来获取输入文件每列中每个值的计数（相对于每列的计数），如果计数大于包含列、值、计数的指定x，则构建一个查找文件。
mapper：使用perl散列来构建包含列、值、计数的散列，其中计数随着该列的值的每次出现而递增：

$ColValCntHash{ $col }{ $value }++;

由于这是使用hadoop流的Map器，因此我以以下键值格式打印结果，以允许对还原器进行更分布式的分配：

col|value\tcount

这将生成列#和字段值的聚合键，然后将该列中该值的计数指定为键。
reducer：读取传递给它的col | value\t计数，解析键，构建一个散列，将从多个Map器传入的值的计数相加。

$FinalCountHash{ $col }{ $value } += $count;

最后，如果col、value和count小于指定的x，则每个reducer都会将col、value和count打印到输出文件中。
第一次传递结果：传递1生成一个“查找”列表，该列表源于传递2中发生消隐的位置，其结构如下：

Col\tValue\tCount
1\t123\t3
3\t234\t2

通过2

这个过程使用hadoop流，只有一个perlMap器（没有reduce阶段）来清空输入文件中的值（如果它们存在于pass1生成的查找文件中）。
Map器：Map器要求将查找文件与分发到每个节点的perl代码一起传送，以便使用它构建查找哈希，如：

$aggKey = $col . "|" . $value;
$LookupHash{ $aggKey }=1;

aggkey用于减少散列的内存开销，现在只是一个散列，它将我可以缓冲到内存中的值的数量增加到大约22 mil。
当输入记录通过Map器传递时，将对列进行迭代，并检查散列以查看列和值是否存在。如果是，则该列中的值将替换为空白。
pass2输出：pass2生成一个输出文件（部件文件），其中包含输入文件的“匿名”版本。

发行

即使集群有足够的内存（2.4tb的总和），这也是在容器之间划分的，在容器出现内存问题之前，pass2最多可以从查找文件中加载大约22 mil的值。这需要对输入进行迭代消隐，方法是将查找文件拆分为22mil块，并在pass2的多个过程中对主文件运行它们。10-100 gb文件的整个过程在5到20分钟内运行，具体取决于要清空的值的数量。
现在我有了一个1.4tb大小的文件来处理这个文件，几乎有20k列，结果查找文件包含46亿条记录（大约是文件中总值的1%，相对于以前的文件是%）。
由于内存限制，一次只能加载22密耳的查找值。。。这使得我当前的进程需要在pass2210次中传递输入文件。。由于它必须读取1.4tb的文件（每次删除都会略微收缩），每次读取都需要25分钟。
我希望这能充分描述问题/当前解决方案/问题。任何帮助都将不胜感激。
谢谢！