jq可以跨文件进行聚合吗

Question

我正在尝试确定一个程序/软件，它可以让我有效地获取大量大型 CSV 文件（总计 40+ GB）并输出具有我需要导入 Elasticsearch (ES) 的特定格式的 JSON 文件.

jq 可以像这样高效地获取数据吗：

file1:
id,age,gender,wave
1,49,M,1
2,72,F,0

file2:
id,time,event1
1,4/20/2095,V39
1,4/21/2095,T21
2,5/17/2094,V39

按 id 聚合它（这样多个文件中 CSV 行中的所有 JSON 文档都属于一个 id 条目），输出如下内容：

{"index":{"_index":"forum_mat","_type":"subject","_id":"1"}}
{"id":"1","file1":[{"filen":"file1","id":"1","age":"49","gender":"M","wave":"1"}],"file2":[{"filen":"file2","id":"1","time":"4/20/2095","event1":"V39"},{"filen":"file2","id":"1","time":"4/21/2095","event1":"T21"}]}
{"index":{"_index":"forum_mat","_type":"subject","_id":"2"}}
{"id":"2","file1":[{"filen":"file1","id":"2","age":"72","gender":"F","wave":"0"}],"file2":[{"filen":"file2","id":"2","time":"5/17/2094","event1":"V39"}]}

我在 Matlab 中编写了一个脚本，但我担心它会很慢。我可能需要几个月的时间来处理所有 40+GB 的数据。我是informed，Logstash（这是 ES 的首选数据输入工具）不擅长这种类型的聚合。

Answer 1

我相信以下内容可以满足您的要求，但我不完全理解您的输入文件与您包含的输出之间的联系。希望这至少能让你走上正轨。

程序假定您的所有数据都可以放入内存。它使用 JSON 对象作为字典进行快速查找，因此性能应该相当不错。

这里采用的方法将 csv-to-json 转换与聚合分开，因为前者可能有更好的方法。 （例如参见the jq Cookbook entry on convert-a-csv-file-with-headers-to-json。）

第一个文件 (scsv2json.jq) 用于将简单的 CSV 转换为 JSON。第二个文件 (aggregate.jq) 进行聚合。有了这些：

$ (jq -R -s -f scsv2json.jq file1.csv ;\ jq -R -s -f scsv2json.jq file2.csv) |\ jq -s -c -f aggregate.jq [{"id":"1", "file1":{"age":"49","gender":"M","wave":"1"}, "file2":{"time":"4/21/2095","event1":"T21"}}, {"id":"2", "file1":{"age":"72","gender":"F","wave":"0"}, "file2":{"time":"5/17/2094","event1":"V39"}}]

请注意，“id”已从输出的内部对象中删除。

聚合.jq：

# Input: an array of objects, each with an "id" field
# such that (tostring|.id) is an index.
# Output: a dictionary keyed by the id field.
def todictionary:
  reduce .[] as $row ( {}; . + { ($row.id | tostring): $row } );

def aggregate:
  .[0] as $file1
  | .[1] as $file2
  | ($file1 | todictionary) as $d1
  | ($file2 | todictionary) as $d2
  | ( [$file1[].id] + [$file2[].id] | unique ) as $keys
  | reduce ($keys[] | tostring) as $k
      ( [];
        . + [{"id": $k, 
              "file1": ($d1[$k] | del(.id)),
              "file2": ($d2[$k] | del(.id)) }] );

aggregate

scsv2json.jq

def objectify(headers):
  . as $in
  | reduce range(0; headers|length) as $i
      ({}; .[headers[$i]] = ($in[$i]) );

def csv2table:
  def trim: sub("^ +";"") |  sub(" +$";"");
  split("\n") | map( split(",") | map(trim) );

def csv2json:
  csv2table
  | .[0] as $headers
  | reduce (.[1:][] | select(length > 0) ) as $row
      ( []; . + [ $row|objectify($headers) ]);

csv2json

以上假设正在使用支持正则表达式的 jq 版本。如果您的 jq 不支持正则表达式，只需省略修剪即可。

Answer 2

这是一种占用内存较少的方法。它只需要 file1 保存在内存中：第二个文件一次处理一行。

调用是这样的：

$ jq -n -R --argfile file1 <(jq -R -s -f scsv2json.jq file1.csv)\
     -f aggregate.jq file2.csv

其中 scsv2json.jq 如上一篇文章所示。此处不再赘述，主要是因为（如其他地方所指出的）其他一些以相同方式将 CSV 转换为 JSON 的程序可能是合适的。

聚合.jq：

def objectify(headers):
  . as $in
  | reduce range(0; headers|length) as $i
      ({}; .[headers[$i]] = ($in[$i]) );

def csv2table:
  def trim: sub("^ +";"") |  sub(" +$";"");
  split("\n") | map( split(",") | map(trim) );

# Input: an array of objects, each with an "id" field
# such that (tostring|.id) is an index.
# Output: a dictionary keyed by the id field.
def todictionary:
  reduce .[] as $row ( {}; . + { ($row.id | tostring): $row } );

# input: {"id": ID } + OBJECT2
# dict: {ID: OBJECT1, ...}
# output: {id: ID, "file1": OBJECT1, "file2": OBJECT2}
def aggregate(dict):
  .id as $id
  | (dict[$id] | del(.id)) as $o1
  | {"id": $id,
     "file1": $o1,
     "file2":  del(.id) };

# $file1 is the JSON version of file1.csv -- an array of objects
(input | csv2table[0]) as $headers
| inputs
| csv2table[0]
| objectify($headers) 
| ($file1 | todictionary) as $d1
| aggregate($d1)

Answer 3

这是一种 jq 内存要求非常小的方法。它假定您已经能够将所有 .csv 文件合并到一个 JSON 数组流（或文件）中，格式如下：

[id, sourceFile, baggage]

其中 id 的值按排序顺序排列。流可能如下所示：

 [1,"file1", {"a":1}]
 [1,"file2", {"b":1}]
 [1,"file3", {"c":1}]
 [2,"file1", {"d":1}]
 [2,"file2", {"e":1}]
 [3,"file1", {"f":1}]

此初步步骤需要全局排序，因此您可能需要仔细选择排序实用程序。

可以有任意多个文件源；不需要每个数组都适合一行；并且 id 值不必是整数——例如，它们可以是字符串。

假设以上内容位于名为 combine.json 的文件中，并且该 aggregate.jq 具有如下所示的内容。然后调用：

$ jq -c -n -f aggregate.jq combined.json

会产生：

{"id":1,"file1":{"a":1},"file2":{"b":1},"file3":{"c":1}}
{"id":2,"file1":{"d":1},"file2":{"e":1}}
{"id":3,"file1":{"f":1}}

更正：聚合.jq：

foreach (inputs,null) as $row
  # At each iteration, if .emit then emit it
  ( {"emit": null, "current": null};

    if $row == null
    then {emit: .current, current: null}          # signal EOF
    else  {id: $row[0], ($row[1]) : $row[2] } as $this
    | if .current == null
      then {emit: null, current: $this}
      elif $row[0] == .current.id
      then .emit = null | .current += $this
      else {emit: .current, current: $this}
      end
    end;
    if .emit then .emit else empty end
  )

Answer 4

正如其中一个 cmets 中所建议的，我最终使用 SQL 以我需要的格式导出 JSON。另一个thread 帮助很大。最后，我选择将给定的 SQL 表输出到它自己的 JSON 文件而不是组合它们（文件大小变得难以管理）。这是执行此操作的代码结构，以便您为 Bulk API 和 JSON 数据行生成命令行：

create or replace function format_data_line(command text, data_str text)
returns setof text language plpgsql as $$
begin
    return next command;
    return next             
        replace(
            regexp_replace(data_str,
                '(\d\d\d\d-\d\d-\d\d)T', '\1 ', 'g'),
            e' \n ', '');
end $$;

COPY (
    with f_1 as(
       SELECT id, json_agg(fileX.*) AS tag
       FROM forum.file3
       GROUP BY id
    )
    SELECT 
        format_data_line(
            format('{"update":{"_index":"forum2","_type":"subject","_id":%s}}',a.id),
            format('{"doc":{"id":%s,"fileX":%s}}', 
                a.id, a.tag))
    FROM f_1 a 
) TO '/path/to/json/fileX.json';

使用 Bulk API 导入较大的文件也被证明是有问题的（内存不足 Java 错误），因此需要一个脚本在给定时间仅将数据的子集发送到 Curl（用于 Elasticsearch 中的索引）。该脚本的基本结构是：

#!/bin/bash

FILE=$1
INC=100
numline=`wc -l $FILE | awk '{print $1}'`
rm -f output/$FILE.txt
for i in `seq 1 $INC $numline`; do
    TIME=`date +%H:%M:%S`
    echo "[$TIME] Processing lines from $i to $((i + INC -1))"
    rm -f intermediates/interm_file_$i.json
    sed -n $i,$((i +INC - 1))p $FILE >> intermediates/interm_file_$i.json
    curl -s -XPOST localhost:9200/_bulk --data-binary @intermediates/interm_file_$i.json >> output/$FILE.txt
done

应该在脚本文件目录下创建一个“中间体”目录。该脚本可以保存为“ESscript”并在命令行上运行：

./ESscript fileX.json