while 循环与来自拆分文件的输入并行答案

【问题标题】：while loops in parallel with input from splited filewhile 循环与来自拆分文件的输入并行
【发布时间】：2019-05-21 00:17:13
【问题描述】：

我坚持这一点。因此，我的代码中有这个 while-read 循环需要很长时间，我想在许多处理器中运行它。但是，我想拆分输入文件并并行运行 14 个循环（因为我有 14 个线程），每个拆分文件一个循环。问题是我不知道如何告诉 while 循环获取和使用哪个文件。

例如，在常规的 while-read 循环中，我会编写代码：

while read line
do
   <some code>
done < input file or variable...

但在这种情况下，我想将上述输入文件拆分为 14 个文件并并行运行 14 个 while 循环，每个拆分文件一个。我试过了：

split -n 14 input_file
find . -name "xa*" | \
        parallel -j 14 | \
        while read line
        do
        <lot of stuff>
        done

也试过

split -n 14 input_file
function loop {
            while read line
            do
                <lot of stuff>
            done
}
export -f loop
parallel -j 14 ::: loop

但我都无法判断哪个文件将是循环的输入，因此并行会理解“将每个 xa* 文件并行放入单独的循环中”

输入文件示例（字符串列表）

AEYS01000010.10484.12283
CVJT01000011.50.2173
KF625180.1.1799
KT949922.1.1791
LOBZ01000025.54942.57580

编辑

这是代码。输出是一个表格（741100 行），其中包含一些关于已经进行的 DNA 序列比对的统计数据。该循环采用带有 DNA 序列继承的 input_file（无虚线，从 500 到 ~45000 行，800Kb），逐行读取并在数据库中查找这些继承的每个对应的完整分类法（~45000 行） .然后，它会进行一些求和/除法。输出是一个 .tsv，如下所示（序列“KF625180.1.1799”的示例）：

Rate of taxonomies for this sequence in %:        KF625180.1.1799 D_6__Bacillus_atrophaeus
Taxonomy %aligned number_ocurrences_in_the_alignment     num_ocurrences_in_databank    %alingment/databank
D_6__Bacillus_atrophaeus   50%     1       20      5%
D_6__Bacillus_amyloliquefaciens    50%     1       154     0.649351%



$ head input file  
AEYS01000010.10484.12283
CVJT01000011.50.217
KF625180.1.1799
KT949922.1.1791
LOBZ01000025.54942.57580

循环内部还使用了两个附加文件。它们不是循环输入。 1) 一个名为 alnout_file 的文件，它仅用于查找给定序列对数据库的命中数（或比对数）。它以前也是在这个循环之外制作的。它的行数可以从数百到数千不等。这里只有第 1 列和第 2 列很重要。 Column1 是序列的名称，col2 是它在数据库中匹配的所有序列的名称。看起来是这样的：

$ head alnout_file
KF625180.1.1799 KF625180.1.1799 100.0   431     0       0       1       431     1       431     -1      0
KF625180.1.1799 KP143082.1.1457 99.3    431     1       2       1       431     1       429     -1      0
KP143082.1.1457 KF625180.1.1799 99.3    431     1       2       1       429     1       431     -1      0

2) 一个数据库 .tsv 文件，其中包含与 DNA 序列相对应的约 45000 个分类。每个分类都在一行中：

$ head taxonomy.file.tsv
KP143082.1.1457 D_0__Bacteria;D_1__Firmicutes;D_2__Bacilli;D_3__Bacillales;D_4__Bacillaceae;D_5__Bacillus;D_6__Bacillus_amyloliquefaciens
KF625180.1.1799 D_0__Bacteria;D_1__Firmicutes;D_2__Bacilli;D_3__Bacillales;D_4__Bacillaceae;D_5__Bacillus;D_6__Bacillus_atrophaeus

所以，给定序列 KF625180.1.1799。我之前将它与一个包含约 45000 个其他 DNA 序列的数据库进行了比对，并得到了一个输出，其中包含它匹配的所有序列的加入。循环的作用是找到所有这些序列的分类并计算我之前提到的“统计数据”。代码为我拥有的所有 DNA 序列加入。

TAXONOMY=path/taxonomy.file.tsv
while read line
do
#find hits
        hits=$(grep $line alnout_file | cut -f 2)
        completename=$(grep $line $TAXONOMY | sed 's/D_0.*D_4/D_4/g')
        printf "\nRate of taxonomies for this sequence in %%:\t$completename\n"
        printf "Taxonomy\t%aligned\tnumber_ocurrences_in_the_alignment\tnum_ocurrences_in_databank\t%alingment/databank\n"

        #find hits and calculate the frequence (%) of the taxonomy in the alignment output
        # ex.: Bacillus_subtilis 33
        freqHits=$(grep "${hits[@]}" $TAXONOMY | \
                cut -f 2 | \
                awk '{a[$0]++} END {for (i in a) {print i, "\t", a[i]/NR*100, "\t", a[i]}}' | \
                sed -e 's/D_0.*D_5/D_5/g' -e 's#\s\t\s#\t#g' | \
                sort -k2 -hr)

        # print frequence of each taxonomy in the databank

        freqBank=$(while read line; do grep -c "$line" $TAXONOMY; done < <(echo "$freqHits" | cut -f 1))
        #print cols with taxonomy and calculations
        paste <(printf %s "$freqHits") <(printf %s "$freqBank") | awk '{print $1,"\t",$2"%","\t",$3,"\t",$4,"\t",$3/$4*100"%"}'

done < input_file

它需要大量的 grep 和解析，因此在一个处理器中运行大约需要 12 小时来处理所有 45000 个 DNA 序列加入。我想拆分 input_file 并在我拥有的所有处理器（14）中执行此操作，因为它会花费时间。谢谢大家对我这么有耐心=)

【问题讨论】：

为什么是14？数据是否分段？或者多个实例可以从 FIFO 中读取它的行吗？
就因为我有14个线程。正在编辑它
soo.... 数据分段了吗？或者多个实例可以从 FIFO 中读取它的行吗？
我不知道你所说的“我有 14 个线程”是什么意思——你需要稍微详细说明一下。
当 grepping 分类中的全名时，我猜你只期望得到 1 个答案，所以在第一场比赛后退出 (grep -m1 ...)，不要费心解析其余的。

标签： bash while-loop parallel-processing

【解决方案1】：

希望这个小脚本能帮到你：

function process {
    while read line; do
        echo "$line"
    done < $1
}

function loop {
    file=$1
    chunks=$2
    dir=`mktemp -d`
    cd $dir
    split -n l/$chunks $file
    for i in *; do
        process "$i" &
    done
    rm -rf $dir
}

loop /tmp/foo 14

它在指定文件上以指定数量的块（不分割行）并行运行进程循环（使用& 将每个调用置于后台）。我希望它能让你开始。

【讨论】：

我认为这是可行的，但不知何故找不到我的文件。我用 $1 替换了 input_file 并且没有用。也替换为 3 美元（因为我在 sdript.sh 前面有 2 个标准输入输入）并且也没有找到该文件。
split：无法打开“input_file”进行阅读：没有这样的文件或目录./script.sh：第50行：input_file：没有这样的文件或目录

【解决方案2】：

这可以为你完成这项工作，我不熟悉并行而不是使用本机 bash 生成进程&：

function loop () {
  while IFS= read -r -d $'\n'
  do
    # YOUR BIG STUFF
  done < "${1}"
}

arr_files=(./xa*)
for i in "${arr_files[@]}"
do loop "${i}" &
done
wait

【讨论】：

【解决方案3】：

作为替代方案，我进行了快速测试。

#! /bin/env bash
mkfifo PIPELINE             # create a single queue
cat "$1" > PIPELINE &       # supply it with records
{ declare -i cnt=0 max=14
  while (( ++cnt <= max ))  # spawn loop creates worker jobs
  do printf -v fn "%02d" $cnt
     while read -r line     # each work loop reads common stdin...
     do echo "$fn:[$line]"
        sleep 1
     done >$fn.log 2>&1 &   # these run in background in parallel
  done                      # this one exits
} < PIPELINE                # *all* read from the same queue
wait
cat [0-9][0-9].log

不需要split，但需要mkfifo。

显然，更改内部循环内的代码。

【讨论】：

【解决方案4】：

这回答了您的问题，即如何并行处理您从运行split 获得的 14 个文件。但是，我认为这不是您想要做的任何事情的最佳方式 - 但我们需要您为此提供一些答案。

所以，让我们制作一百万行文件并将其拆分为 14 部分：

seq 1000000 > 1M
split -n 14 1M part-

这给了我 14 个名为 part-aa 到 part-an 的文件。现在您的问题是如何并行处理这 14 个部分 - （先阅读最后一行）：

#!/bin/bash

# This function will be called for each of the 14 files
DoOne(){
   # Pick up parameters
   job=$1
   file=$2
   # Count lines in specified file
   lines=$(wc -l < "$file")
   echo "Job No: $job, file: $file, lines: $lines"
}

# Make the function above known to processes spawned by GNU Parallel
export -f DoOne

# Run 14 parallel instances of "DoOne" passing job number and filename to each
parallel -k -j 14 DoOne {#} {} ::: part-??

样本输出

Job No: 1, file: part-aa, lines:    83861
Job No: 2, file: part-ab, lines:    72600
Job No: 3, file: part-ac, lines:    70295
Job No: 4, file: part-ad, lines:    70295
Job No: 5, file: part-ae, lines:    70294
Job No: 6, file: part-af, lines:    70295
Job No: 7, file: part-ag, lines:    70295
Job No: 8, file: part-ah, lines:    70294
Job No: 9, file: part-ai, lines:    70295
Job No: 10, file: part-aj, lines:    70295
Job No: 11, file: part-ak, lines:    70295
Job No: 12, file: part-al, lines:    70294
Job No: 13, file: part-am, lines:    70295
Job No: 14, file: part-an, lines:    70297

您通常会省略 GNU Parallel 的 -k 参数 - 我只是添加它以便输出按顺序排列。

【讨论】：

【解决方案5】：

您正在寻找--pipe。在这种情况下你甚至可以使用优化的--pipepart（版本>20160621）：

export TAXONOMY=path/taxonomy.file.tsv
doit() {
while read line
do
#find hits
        hits=$(grep $line alnout_file | cut -f 2)
        completename=$(grep $line $TAXONOMY | sed 's/D_0.*D_4/D_4/g')
        printf "\nRate of taxonomies for this sequence in %%:\t$completename\n"
        printf "Taxonomy\t%aligned\tnumber_ocurrences_in_the_alignment\tnum_ocurrences_in_databank\t%alingment/databank\n"

        #find hits and calculate the frequence (%) of the taxonomy in the alignment output
        # ex.: Bacillus_subtilis 33
        freqHits=$(grep "${hits[@]}" $TAXONOMY | \
                cut -f 2 | \
                awk '{a[$0]++} END {for (i in a) {print i, "\t", a[i]/NR*100, "\t", a[i]}}' | \
                sed -e 's/D_0.*D_5/D_5/g' -e 's#\s\t\s#\t#g' | \
                sort -k2 -hr)

        # print frequence of each taxonomy in the databank

        freqBank=$(while read line; do grep -c "$line" $TAXONOMY; done < <(echo "$freqHits" | cut -f 1))
        #print cols with taxonomy and calculations
        paste <(printf %s "$freqHits") <(printf %s "$freqBank") | awk '{print $1,"\t",$2"%","\t",$3,"\t",$4,"\t",$3/$4*100"%"}'

done
}
export -f doit
parallel -a input_file --pipepart doit

这会将 input_file 分成 10*ncpu 块（其中 ncpu 是 CPU 线程数），将每个块传递给 doit，并行运行 ncpu 作业。

也就是说，我认为你真正的问题是产生了太多的程序：如果你用 Perl 或 Python 重写 doit，我希望你会看到一个显着的加速。

【讨论】：

这行得通，但我正在听从你和其他人的建议，并为它编写一个 Python 代码。谢谢

【解决方案6】：

我认为在这里使用一堆 grep 和 awk 命令是错误的方法 - 使用 Perl 或 awk 会更好。由于您没有提供任何示例文件，因此我使用此代码生成了一些示例文件：

#!/bin/bash

for a in {A..Z} {0..9} ; do
   for b in {A..Z} {0..9} ; do
      for c in {A..Z} {0..9} ; do
         echo "${a}${b}${c}"
      done
   done
done > a

# Now make file "b" which has the same stuff but shuffled into a different order
gshuf < a > b

请注意，字母表中有 26 个字母，所以如果我将数字 0..9 添加到字母表的字母中，我会得到 36 个字母数字数字，如果我嵌套 3 个循环，我会得到 36^3 或 46,656与您的文件大小大致匹配的行。文件 a 现在看起来像这样：

AAA
AAB
AAC
AAD
AAE
AAF

文件b 如下所示：

UKM
L50
AOC
79U
K6S
6PO
12I
XEV
WJN

现在我想循环遍历a 在b 中找到对应的行。首先，我使用你的方法：

time while read thing ; do grep $thing b > /dev/null ; done < a

这需要 9 分 35 秒。

如果我现在在第一场比赛中退出grep，平均我会在中间找到它，这意味着时间将减半，因为我不会在找到我想要的东西后继续不必要地阅读b .

time while read thing ; do grep -m1 $thing b > /dev/null ; done < a

这将时间缩短到 4 分 30 秒。

如果我现在使用awk 将b 的内容读入关联数组（也称为哈希），然后读取a 的元素并在b 中找到它们，如下所示：

time awk 'FNR==NR{a[$1]=$1; next} {print a[$1]}' b a > /dev/null

现在运行时间为 0.07 秒。希望你明白我在做什么。我希望 Perl 会同时做到这一点，并且还为循环中间的数学提供更多表达能力。

【讨论】：

我正在遵循建议并打算在 Python 中实现它。感谢您的建议
Python 也应该简单地做到这一点。祝你好运！完成后将您的解决方案发回与 SO 社区分享，并接受您自己的解决方案以获取积分。