【问题标题】:Combining big data files with different columns into one big file将不同列的大数据文件合并为一个大文件
【发布时间】:2020-03-10 04:50:34
【问题描述】:

我有 N 个制表符分隔的文件。每个文件都有一个标题行,说明列的名称。有些列对所有文件都是通用的,但有些列是唯一的。

我想将所有文件合并为一个包含所有相关标题的大文件。

例子:

> cat file1.dat
a b c
5 7 2
3 9 1

> cat file2.dat
a b e f
2 9 8 3
2 8 3 3
1 0 3 2

> cat file3.dat
a c d g
1 1 5 2

> merge file*.dat
a b c d e f g
5 7 2 - - - -
3 9 1 - - - -
2 9 - - 8 3 -
2 8 - - 3 3 -
1 0 - - 3 2 -
1 - 1 5 - - 2

- 可以替换为任何内容,例如 NA。

警告:文件太大,我无法同时将它们全部加载到内存中。

我在 R 中有一个使用

的解决方案
write.table(do.call(plyr:::rbind.fill, 
            Map(function(filename) 
                    read.table(filename, header=1, check.names=0), 
                filename=list.files('.'))), 
    'merged.dat', quote=FALSE, sep='\t', row.names=FALSE)

但是当数据太大时,这会失败并出现内存错误。

实现这一目标的最佳方法是什么?

我认为最好的方法是首先遍历所有文件以收集列名,然后遍历文件以将它们放入正确的格式,并在遇到它们时将它们写入磁盘。但是,是否可能已经有一些可用的代码可以执行此操作?

【问题讨论】:

  • “这么大”有多大? GB、TB、ZB?
  • @jhnc 这次是几个 TB
  • 合并文件中是否应该有一个“d”列?没有一个文件有这样的列。
  • @Danny_ds 抱歉,这是个错误,已修复!
  • @rhombidodedecahedron:你应该使用每个答案来分析性能,这样我们也可以知道哪个答案最好

标签: r bash dataframe awk


【解决方案1】:

从算法的角度来看,我会采取以下步骤:

  1. 处理标题:

    • 读取所有输入文件的所有标题并提取所有列名
    • 按您想要的顺序对列名进行排序
    • 创建一个查找表,当给定字段号时返回列名 (h[n] -> "name")
  2. 处理文件:在标题之后,您可以重新处理文件

    • 读取文件头
    • 创建一个查找表,在给定列名时返回字段编号。关联数组在这里很有用:(a["name"] -> field_number)
    • 处理文件的其余部分

      1. 遍历合并文件的所有字段
      2. 用h获取列名
      3. 检查列名是否在a,如果不在则打印-,如果是则打印a对应的字段号。

这很容易通过使用扩展名nextfile 和asorti 的GNU awk 完成。 nextfile 函数允许我们只读取标题并移动到下一个文件而不处理整个文件。由于我们需要处理文件两次(第 1 步读取标题和第 2 步读取文件),我们将要求 awk 动态操作其参数列表。每次处理文件头时,我们将它添加到参数列表的末尾ARGV,以便它可以用于step 2。

BEGIN { s="-" }                # define symbol
BEGIN { f=ARGC-1 }             # get total number of files
f { for (i=1;i<=NF;++i) h[$i]  # read headers in associative array h[key]
    ARGV[ARGC++] = FILENAME    # add file at end of argument list
    if (--f == 0) {            # did we process all headers?
       n=asorti(h)             # sort header into h[idx] = key
       for (i=1;i<=n;++i)      # print header
           printf "%s%s", h[i], (i==n?ORS:OFS)
    }
    nextfile                   # end of processing headers
}           
# Start of processing the files
(FNR==1) { delete a; for(i=1;i<=NF;++i) a[$i]=i; next } # read header
{ for(i=1;i<=n;++i) printf "%s%s", (h[i] in a ? $(a[h[i]]) : s), (i==n?ORS:OFS) }

如果您将上述内容存储在文件merge.awk 中,您可以使用以下命令:

awk -f merge.awk f1 f2 f3 f4 ... fx

类似的方式,但使用f 时不那么麻烦:

BEGIN { s="-" }                 # define symbol
BEGIN {                         # modify argument list from
        c=ARGC;                 #   from: arg1 arg2  ... argx
        ARGV[ARGC++]="f=1"      #   to:   arg1 arg2  ... argx f=1 arg1 arg2  ... argx
        for(i=1;i<c;++i) ARGV[ARGC++]=ARGV[i]
}
!f { for (i=1;i<=NF;++i) h[$i]  # read headers in associative array h[key]
     nextfile
}
(f==1) && (FNR==1) {            # process merged header
     n=asorti(h)                # sort header into h[idx] = key
     for (i=1;i<=n;++i)         # print header
        printf "%s%s", h[i], (i==n?ORS:OFS)
     f=2                         
}
# Start of processing the files
(FNR==1) { delete a; for(i=1;i<=NF;++i) a[$i]=i; next } # read header
{ for(i=1;i<=n;++i) printf "%s%s", (h[i] in a ? $(a[h[i]]) : s), (i==n?ORS:OFS) }

此方法略有不同,但允许将具有不同字段分隔符的文件处理为

awk -f merge.awk f1 FS="," f2 f3 FS="|" f4 ... fx

如果您的参数列表太长,您可以使用awk 为您创建它:

BEGIN { s="-" }                 # define symbol
BEGIN {                         # read argument list from input file:
  fname=(ARGC==1 ? "-" : ARGV[1])
  ARGC=1                        # from: filelist or /dev/stdin
  while ((getline < fname) > 0) #   to:   arg1 arg2 ... argx
     ARGV[ARGC++]=$0
}
BEGIN {                         # modify argument list from
        c=ARGC;                 #   from: arg1 arg2  ... argx
        ARGV[ARGC++]="f=1"      #   to:   arg1 arg2  ... argx f=1 arg1 arg2  ... argx
        for(i=1;i<c;++i) ARGV[ARGC++]=ARGV[i]
}
!f { for (i=1;i<=NF;++i) h[$i]  # read headers in associative array h[key]
     nextfile
}
(f==1) && (FNR==1) {            # process merged header
     n=asorti(h)                # sort header into h[idx] = key
     for (i=1;i<=n;++i)         # print header
        printf "%s%s", h[i], (i==n?ORS:OFS)
     f=2                         
}
# Start of processing the files
(FNR==1) { delete a; for(i=1;i<=NF;++i) a[$i]=i; next } # read header
{ for(i=1;i<=n;++i) printf "%s%s", (h[i] in a ? $(a[h[i]]) : s), (i==n?ORS:OFS) }

可以这样运行:

$ awk -f merge.awk filelist
$ find . | awk -f merge.awk "-"
$ find . | awk -f merge.awk

或任何类似的命令。

如您所见,通过仅添加一小段代码,我们能够灵活地调整为 awk 代码来支持我们的需求。

【讨论】:

  • 花了一分钟的时间来了解f/ARGV/ARGC 代码...您将每个输入文件动态添加回awk 的输入队列,例如,awk -f merge.awk f1 f2 f3 变为awk -f merge.awk f1 f2 f3 f1 f2 f3 ...整洁的;对不起,我不能给你超过 +1 ! ;是的,你在描述中说了很多,但直到我理解了f/ARGV/ARGC 代码才点击它:-)
  • @markp 我对文本进行了一些更新以使其清楚。感谢您的支持。
【解决方案2】:

Miller (johnkerl/miller) 在处理大文件时没有得到充分利用。它具有所有有用的文件处理工具中包含的大量功能。就像官方文档说的那样

Miller 类似于 awk, sed, cut, join 和 sort,用于名称索引数据,例如 CSV、TSV 和表格 JSON。您可以使用命名字段来处理数据,而无需计算位置列索引。

对于这种特殊情况,它支持动词unsparsify,文档中说

在所有输入记录上打印具有字段名称并集的记录。 对于给定记录中不存在但存在于其他记录中的字段名称,填写 一个值。这个动词在产生任何输出之前保留所有输入。

您只需要执行以下操作并根据需要将文件重新排序为列位置

mlr --tsvlite --opprint unsparsify then reorder -f a,b,c,d,e,f file{1..3}.dat

一次性产生输出

a   b   c   d   e   f   g
5   7   2   -   -   -   -
3   9   1   -   -   -   -
2   9   -   -   8   3   -
2   8   -   -   3   3   -
1   0   -   -   3   2   -
1   -   1   5   -   -   2

您甚至可以自定义可用于填充空白字段的字符,默认为-。对于自定义字符,请使用 unsparsify --fill-with '#'

所用字段的简要说明

  • 要将输入流分隔为制表符分隔的内容,--tsvlite
  • 漂亮地打印表格数据--opprint
  • unsparsify 就像上面解释的那样对所有输入流中的所有字段名称进行联合
  • 需要重新排序动词reorder,因为列标题在文件之间以随机顺序出现。因此,要明确定义顺序,请将-f 选项与您希望输出显示的列标题一起使用。

软件包的安装非常简单。 Miller 是用可移植的现代 C 语言编写的,具有零运行时依赖性。 installation via package managers 非常简单,它支持所有主要的包管理器 Homebrew、MacPorts、apt-get、apt 和 yum。

【讨论】:

  • wrt reorder -f a,b,c,d,e,f - 通过读取输入文件动态确定顺序是问题的很大一部分。为此,您大概可以使用reorder -f "$(head -1 -q file{1..3}.dat | tr ' ' '\n' | sort -u | paste -sd, -)"
  • 显然 OP 有 10^5 个输入文件,这对于他们正在使用的机器上的 arg 列表来说太多了(请参阅stackoverflow.com/questions/56563626/…)所以我上面关于如何获取字段列表的建议动态不起作用,您需要尝试其他方法...等一下 - 在这种情况下,您无论如何都不能调用 mlr ... file*.dat,因为对于 arg 列表来说文件太多了。
【解决方案3】:

鉴于您在 cmets 中更新了有关拥有大约 10^5 个输入文件的信息(因此超过了非内置命令的 shell 最大参数数)并希望输出列按照它们看到的顺序而不是按字母顺序排序,以下将使用任何 awk 和任何 find 工作:

$ cat tst.sh
#!/bin/env bash
find . -maxdepth 1 -type f -name "$1" |
awk '
NR==FNR {
    fileName = $0
    ARGV[ARGC++] = fileName
    if ( (getline fldList < fileName) > 0 ) {
        if ( !seenList[fldList]++ ) {
            numFlds = split(fldList,fldArr)
            for (inFldNr=1; inFldNr<=numFlds; inFldNr++) {
                fldName = fldArr[inFldNr]
                if ( !seenName[fldName]++ ) {
                    hdr = (numOutFlds++ ? hdr OFS : "") fldName
                    outNr2name[numOutFlds] = fldName
                }
            }
        }
    }
    close(fileName)
    next
}
FNR == 1 {
    if ( !doneHdr++ ) {
        print hdr
    }
    delete name2inNr
    for (inFldNr=1; inFldNr<=NF; inFldNr++) {
        fldName = $inFldNr
        name2inNr[fldName] = inFldNr
    }
    next
}
{
    for (outFldNr=1; outFldNr<=numOutFlds; outFldNr++) {
        fldName = outNr2name[outFldNr]
        inFldNr = name2inNr[fldName]
        fldValue = (inFldNr ? $inFldNr : "-")
        printf "%s%s", fldValue, (outFldNr<numOutFlds ? OFS : ORS)
    }
}
' -

.

$ ./tst.sh 'file*.dat'
a b c e f d g
5 7 2 - - - -
3 9 1 - - - -
2 9 - 8 3 - -
2 8 - 3 3 - -
1 0 - 3 2 - -
1 - 1 - - 5 2

请注意,脚本的输入现在是您希望 find 用来查找文件的通配模式,而不是文件列表。


原答案:

如果您不介意组合的 shell+awk 脚本,那么这将使用任何 awk:

$ cat tst.sh
#!/bin/env bash

awk -v hdrs="$(head -1 -q "$@" | tr ' ' '\n' | sort -u)" '
BEGIN {
    numOutFlds = split(hdrs,outNr2name)
    for (outFldNr=1; outFldNr<=numOutFlds; outFldNr++) {
        fldName = outNr2name[outFldNr]
        printf "%s%s", fldName, (outFldNr<numOutFlds ? OFS : ORS)
    }
}
FNR == 1 {
    delete name2inNr
    for (inFldNr=1; inFldNr<=NF; inFldNr++) {
        fldName = $inFldNr
        name2inNr[fldName] = inFldNr
    }
    next
}
{
    for (outFldNr=1; outFldNr<=numOutFlds; outFldNr++) {
        fldName = outNr2name[outFldNr]
        inFldNr = name2inNr[fldName]
        fldValue = (inFldNr ? $inFldNr : "-")
        printf "%s%s", fldValue, (outFldNr<numOutFlds ? OFS : ORS)
    }
}
' "$@"

.

$ ./tst.sh file{1..3}.dat
a b c d e f g
5 7 2 - - - -
3 9 1 - - - -
2 9 - - 8 3 -
2 8 - - 3 3 -
1 0 - - 3 2 -
1 - 1 5 - - 2

否则这都是使用 GNU awk 处理数组、sorted_in 和 ARGIND 的 awk:

$ cat tst.awk
BEGIN {
    for (inFileNr=1; inFileNr<ARGC; inFileNr++) {
        inFileName = ARGV[inFileNr]
        if ( (getline < inFileName) > 0 ) {
            for (inFldNr=1; inFldNr<=NF; inFldNr++) {
                fldName = $inFldNr
                name2inNr[fldName][inFileNr] = inFldNr
            }
        }
        close(inFileName)
    }

    PROCINFO["sorted_in"] = "@ind_str_asc"
    for (fldName in name2inNr) {
        printf "%s%s", (numOutFlds++ ? OFS : ""), fldName
        for (inFileNr in name2inNr[fldName]) {
            outNr2inNr[numOutFlds][inFileNr] = name2inNr[fldName][inFileNr]
        }
    }
    print ""
}

FNR > 1 {
    for (outFldNr=1; outFldNr<=numOutFlds; outFldNr++) {
        inFldNr = outNr2inNr[outFldNr][ARGIND]
        fldValue = (inFldNr ? $inFldNr : "-")
        printf "%s%s", fldValue, (outFldNr<numOutFlds ? OFS : ORS)
    }
}

.

$ awk -f tst.awk file{1..3}.dat
a b c d e f g
5 7 2 - - - -
3 9 1 - - - -
2 9 - - 8 3 -
2 8 - - 3 3 -
1 0 - - 3 2 -
1 - 1 5 - - 2

为了提高效率,上面的第二个脚本在 BEGIN 部​​分完成了所有繁重的工作,因此在每个输入行评估一次的脚本主体中剩下的工作尽可能少。在 BEGIN 部​​分,它创建一个关联数组 (outNr2inNr[]),将传出字段编号(所有输入文件中所有字段名称按字母顺序排序的列表)映射到传入字段编号,因此在正文中剩下要做的就是打印按该顺序排列的字段。

【讨论】:

  • 感谢您的回复,但是当我执行./tst.sh file*.dat 时出现错误/usr/bin/awk: Argument list too long。有什么想法吗?
  • 您的目录中有多少文件与通配符 file*.dat 匹配?
  • 我希望合并至少 10^5 个文件
【解决方案4】:

这是我(OP)到目前为止提出的解决方案。与其他方法相比,它可能具有一些优势,因为它可以并行处理文件。

R 代码:

library(parallel)
library(parallelMap)

# specify the directory containing the files we want to merge
args <- commandArgs(TRUE)
directory <- if (length(args)>0) args[1] else 'sg_grid'
#output_fname <- paste0(directory, '.dat')

# make a tmp directory that will store all the files 
tmp_dir <- paste0(directory, '_tmp')
dir.create(tmp_dir)

# list the .dat files we want to merge 
filenames <- list.files(directory)
filenames <- filenames[grep('.dat', filenames)]

# a function to read the column names 
get_col_names <- function(filename) 
    colnames(read.table(file.path(directory, filename), 
        header=T, check.names=0, nrow=1))

# grab all the headers of all the files and merge them 
col_names <- get_col_names(filenames[1])
for (simulation in filenames) {
    col_names <- union(col_names, get_col_names(simulation))
}

# put those column names into a blank data frame 
name_DF <- data.frame(matrix(ncol = length(col_names), nrow = 0))
colnames(name_DF) <- col_names

# save that as the header file 
write.table(name_DF, file.path(tmp_dir, '0.dat'), 
    col.names=TRUE, row.names=F, quote=F, sep='\t')

# now read in every file and merge with the blank data frame 
# it will have NAs in any columns it didn't have before 
# save it to the tmp directory to be merged later 
parallelStartMulticore(max(1, 
    min(as.numeric(Sys.getenv('OMP_NUM_THREADS')), 62)))
success <- parallelMap(function(filename) {
    print(filename)
    DF <- read.table(file.path(directory, filename), 
        header=1, check.names=0)
    DF <- plyr:::rbind.fill(name_DF, DF)
    write.table(DF, file.path(tmp_dir, filename), 
        quote=F, col.names=F, row.names=F, sep='\t')
}, filename=filenames)

# and we're done 
print(all(unlist(success)))

这会创建所有文件的临时版本,现在每个文件都有所有标题,然后我们可以将它们 cat 一起放入结果中:

ls -1 sg_grid_tmp/* | while read fn ; do cat "$fn" >> sg_grid.dat; done

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2021-05-28
    • 1970-01-01
    • 2020-12-26
    • 2021-12-23
    • 2019-11-06
    • 2012-05-31
    • 1970-01-01
    相关资源
    最近更新 更多