嵌套 GNU Parallel 处理多个大文件并将每个文件数据拆分为队列处理

Posted 2023-03-24

【中文标题】嵌套 GNU Parallel 处理多个大文件并将每个文件数据拆分为队列处理

【英文标题】：nesting GNU Parallel to process multiple huge files and split each file data to be processed as queue

【发布时间】：2019-03-29 14:57:30

【问题描述】：

我有一个包含近 100 个日志文件的目录，每个日志文件的大小为 10~15 GB。要求是逐行读取每个文件（顺序根本不重要），清理行json并将其转储到后端elasticsearch存储进行索引。

这是我做这项工作的工人

# file = worker.php

echo " -- New PHP Worker Started -- "; // to get how many times gnu-parallel initiated the worker
$dataSet = [];

while (false !== ($line = fgets(STDIN))) 

    // convert line text to json
    $l = json_decode($line);
    $dataSet[] = $l;

    if(sizeof($dataSet) >= 1000) 
        //index json to elasticsearch
        $elasticsearch->bulkIndex($dataSet);
        $dataSet = []; 
    

在here 和here 答案的帮助下，我快到了，它正在工作（有点），但只需要确保它实际上在做我假设它正在做的事情.

只需一个文件，我就可以按如下方式处理它

parallel --pipepart -a 10GB_input_file.txt  --round-robin php worker.php 

效果很好。添加 --round-robin 确保 php 工作进程只启动一次，然后它只是继续接收数据作为管道（可怜的人的队列）。

所以对于 4CPU 的机器，它会启动 4 个 php worker 并非常快速地处理所有数据。

要对所有文件执行相同操作，这是我的看法

find /data/directory -maxdepth 1 -type f | parallel cat | parallel --pipe -N10000 --round-robin php worker.php 

这有点像工作，但我有一种直觉，这是对所有文件并行嵌套的错误方法。

其次，由于它不能使用--pipepart，我认为它比较慢。

第三，一旦工作完成，我看到在 4cpu 的机器上，只启动了 4 个工人并完成了工作。这是正确的行为吗？它不应该为每个文件启动 4 个工作人员吗？只是想确保我没有遗漏任何数据。

知道如何以更好的方式做到这一点吗？

【问题讨论】：

你的意思是json_encode()？

【参考方案1】：

如果它们的大小大致相同，为什么不简单地给每个文件一个文件：

find /data/directory -maxdepth 1 -type f |
  parallel php worker.php '<' 

另一种方法是在每个人身上使用--pipepart：

do_one() 
  parallel --pipepart -a "$1" --block -1 php worker.php

export -f do_one
find /data/directory -maxdepth 1 -type f | parallel -j1 do_one

如果启动php worker.php不需要很长时间，那么最后一个可能更可取，因为如果文件大小不同，它会分布得更均匀，因此如果最后一个文件很大，你不要结束等待单个进程完成处理。