はじめに

誰もが知っている通り jq コマンドは JSON データを処理するためのフィルタコマンドです。awk コマンドと同じように抽出や編集といったデータ処理を行える専用の言語を備えています。jq コマンドは巨大な JSON データをストリーミングで処理することができる --stream オプションを持っており、データの完全な取得を待たずにデータを受け取りながら処理することが出来ます。しかしその使い方は難しくあまり解説されていません。そこでどのように使うと良いのかを調べてまとめました。

ストリーミング形式の出力（--stream)

まず次のような JSON データを用意しました。

list.json

[
  {"name": "apple", "price": 210, "count": 10 },
  {"name": "banana", "price": 140, "count": 15 },
  {"name": "orange", "price": 200, "count": 13 }
]

これを --stream オプションを付けてパースすると以下のようになります。

$ jq --stream -c . list.json
[[0,"name"],"apple"]
[[0,"price"],210]
[[0,"count"],10]
[[0,"count"]]            ★
[[1,"name"],"banana"]
[[1,"price"],140]
[[1,"count"],15]
[[1,"count"]]            ★
[[2,"name"],"orange"]
[[2,"price"],200]
[[2,"count"],13]
[[2,"count"]]            ★
[[2]]                    ★

なかなか分かりづらい形式ですがポイントは2つです。

値がキー（パスの配列表現）とともに配列にペアで出力されている
オブジェクトまたは配列の終端を意味する値の無い項目（★の部分）が出力されている

検証のための速度制限は pv コマンドで

ストリーミング処理が行われているかの検証には pv コマンドを使うと便利です。パイプ通信の速度を遅くすることが出来ます。

$ pv -q -L 10 list.json | jq --stream -c .
[[0,"name"],"apple"]        【約 4 秒後に出力開始】
[[0,"price"],210]
[[0,"count"],10]
[[0,"count"]]
[[1,"name"],"banana"]       【約 4 秒後に出力開始】
[[1,"price"],140]
[[1,"count"],15]
[[1,"count"]]
[[2,"name"],"orange"]       【約 4 秒後に出力開始】
[[2,"price"],200]
[[2,"count"],13]
[[2,"count"]]
[[2]]

この記事のいくつかの例では（わかりやすく説明する都合上）ストリーミング処理が行われていないものがあります。実際の使用では意図した動作が行われているかを確認してください。

ストリーミング処理の3つの関数

jq コマンドにはストリーミング処理に関連する関数が 3 つ用意されています（参照）

tostream
fromstream(stream_expression)
truncate_stream(stream_expression)

ストリーム形式で出力する (tostream)

これは JSON データをストリーム形式に変換するフィルタのようです。--stream オプションの代わりに使うことができます。

--streamと出力は同じ

$ jq -c 'tostream' list.json
[[0,"name"],"apple"]
[[0,"price"],210]
[[0,"count"],10]
[[0,"count"]]
[[1,"name"],"banana"]
[[1,"price"],140]
[[1,"count"],15]
[[1,"count"]]
[[2,"name"],"orange"]
[[2,"price"],200]
[[2,"count"],13]
[[2,"count"]]
[[2]]

ストリーム形式を元に戻す (fromstream)

fromstream 関数は先程と反対にストリーム形式を入力として元の JSON 形式に戻す関数です。

$ jq -c 'tostream' list.json | jq -cs 'fromstream(.[])'
[
  {"name":"apple","price":210,"count":10},
  {"name":"banana","price":140,"count":15},
  {"name":"orange","price":200,"count":13}
]

【別の書き方】
$ jq -c 'tostream' list.json | jq -cn 'fromstream(inputs)'
[
  {"name":"apple","price":210,"count":10},
  {"name":"banana","price":140,"count":15},
  {"name":"orange","price":200,"count":13}
]

注 出力は見やすく加工しています

次のようなストリーム形式の JSON データを作成し、

list2.json

[[0,"name"],"apple"]
[[0,"price"],210]
[[0,"count"],10]
[[0,"count"]]
[[1,"name"],"banana"]
[[1,"price"],140]
[[1,"count"],15]
[[1,"count"]]
[[3,"name"],"pear"]      ┐
[[3,"price"],220]        │  ここにデータを追加
[[3,"count"],13]         │
[[3,"count"]]            ┘
[[2,"name"],"orange"]
[[2,"price"],200]
[[2,"count"],13]
[[2,"count"]]
[[2]]

jq コマンドに処理させると想定通り本来の形の JSON データに戻すことができました。ちなみに追加したデータを見るとわかるように挿入場所は上記の例のように途中でも良いが同じキーにしてはいけないようです。インデックス番号を 3 ではなく 2 にすると、同じ 2 のデータで上書きされました。

$ jq -cs 'fromstream(.[])' list2.json
[
  {"name":"apple","price":210,"count":10},
  {"name":"banana","price":140,"count":15},
  {"name":"orange","price":200,"count":13}, 
  {"name":"pear","price":220,"count":13}]
]

注 出力は見やすく加工しています

このような変換ができることから awk コマンドなどで行を追加などして JSON データを組み立てることが出来ることがわかります。必ずしも jq コマンドを駆使して JSON データを修正する必要はありません。

配列表現のパスの変更 (truncate_stream)

ドキュメント読んでもなかなかに意味がわかりませんが、以下のようにキーのパス（配列表現）を指定した数だけ頭の方から取り除く関数です。指定した数っていうのが、1 | のようにパイプで渡すのが分かりづらいところです。そのせいで inputs 関数を使わなければならず、-n オプションで最初の一行を読み込まないようにしなければならず、なんとも分かりづらいですね。

$ jq --stream -nc '1 | truncate_stream(inputs)' list.json
[["name"],"apple"]
[["price"],210]
[["count"],10]
[["count"]]
[["name"],"banana"]
[["price"],140]
[["count"],15]
[["count"]]
[["name"],"orange"]
[["price"],200]
[["count"],13]
[["count"]]

【❌ 変数を使うと-nを使わずに書けるがストリーミング処理にならない】
$ jq --stream -cs '. as $inputs | 1 | truncate_stream($inputs.[])' list.json

fromstream の例では配列のキー（インデックス番号 3）を書かなければなりませんでしたが、この形式から元の JSON に戻せば配列のキーのインデックス番号を意識しなくてすみます。それにはこのように書きます。

$ jq --stream -nc '1 | truncate_stream(inputs)' list.json | jq -sc '[fromstream(.[])]'
[
  {"name":"apple","price":210,"count":10},
  {"name":"banana","price":140,"count":15},
  {"name":"orange","price":200,"count":13}
]

【別の書き方】
$ jq --stream -nc '1 | truncate_stream(inputs)' list.json | jq -cn '[fromstream(inputs)]'
[
  {"name":"apple","price":210,"count":10},
  {"name":"banana","price":140,"count":15},
  {"name":"orange","price":200,"count":13}
]

awk につなぐにはどうするか？

ここまでは jq コマンドを使って JSON データを加工する方法でした。実際には JSON データを加工するのではなく読み取ったデータから何らかの処理を行いたいというのが一般的ではないでしょうか？ jq コマンドに強い人であればそれらを含めて全て jq コマンドで行うのでしょうが、他のコマンド（例 awk）につなげて処理する方法を考えてみました。

まず基本的な考え方として awk コマンドで処理しやすいようにストリーム形式を TSV 形式に変換することにします。そのために以下のように書きます。

$ jq --stream -r '[(.[0] | join(".")), .[1]] | @tsv' list.json
0.name	apple
0.price	210
0.count	10
0.count
1.name	banana
1.price	140
1.count	15
1.count
2.name	orange
2.price	200
2.count	13
2.count
2

すぐに気づくと思いますが、パスの配列表現は . 区切りで結合して一つの文字列にしています。データの境目は最初の数値が変わる所で区別することができます。しかしもう少しなんとかしたい所です。そこで truncate_stream を使って数値の部分を取り除きます。

$ jq --stream -r -n '1 | truncate_stream(inputs) | [(.[0] | join(".")), .[1]] | @tsv' list.json
name	apple
price	210
count	10
count
name	banana
price	140
count	15
count
name	orange
price	200
count	13
count

これだと（頑張ればなんとかなると思いますが）データの境目がよくわかりません。そこでデータの境目の項目（値がない行）を空行に変換します。

$ jq --stream -r -n '1 | truncate_stream(inputs) 
    | if length >1 then [(.[0] | join(".")), .[1]] else [] end | @tsv' list.json
name	apple
price	210
count	10

name	banana
price	140
count	15

name	orange
price	200
count	13

この空行で区切られたデータ形式は実は awk コマンドで扱いやすい形式です。通常 awk は一行が一レコードですが、レコードの区切りを改行ではなく空行にすることができます。そのために RS 変数を空にします。次に FS 変数を変更してフィールドの区切りをスペースから改行に変更します。すると以下のように一つのフィールドが「名前＋タブ＋値」形式でデータを読み取ることができます。

$ jq --unbuffered --stream -r -n '1 | truncate_stream(inputs) 
    | if length >1 then [(.[0] | join(".")), .[1]] else [] end | @tsv' list.json \
    | awk -v RS='' -v FS='\n' '{ print $1 " | " $2 " | " $3 }'
name	apple | price	210 | count	10
name	banana | price	140 | count	15
name	orange | price	200 | count	13

ここで --unbuffered を指定している理由は、jq コマンドの出力を端末ではなくパイプで別のコマンドにつなぐと、デフォルトではバッファリングが行われストリーミング処理が滞ってしまうからです。ちなみにバッファリングの挙動は jq コマンドだけではなく多くのコマンドで行われる一般的な挙動で、いくつかのコマンドは同様のオプションを備えています。バッファリングが行われてもストリーミング処理が全く行われなくなるわけではなくデータの入出力速度が十分に速い場合はバッファにすぐに貯まるのでほぼリアルタイムで処理されます。データの入出力速度が速い場合は --unbuffered を指定することで遅くなる場合があるので場合によって使い分ける必要があります。

話を戻して、「名前＋タブ＋値」形式のままでは処理しづらいので扱いやすいようにキーと値に分割して連想配列に入れます。

$ jq --unbuffered --stream -r -n '1 | truncate_stream(inputs) 
    | if length >1 then [(.[0] | join(".")), .[1]] else [] end | @tsv' list.json \
    | awk -v RS='' -v FS='\n' '
      {
        delete fields
        for (i = 1; i<= NF; i++) {
          pos = index($i, "\t")
          fields[substr($i, 1, pos - 1)] = substr($i, pos + 1)
        }
        print fields["name"], fields["price"], fields["count"]
      }
    '
apple 210 10
banana 140 15
orange 200 13

これにて JSON の世界から Unix の世界へと移動が完了です。このようにうまく Unix コマンドにつなげられるのを見ると、jq コマンドはよく出来ているなと思いますね。

さいごに

実際の JSON データはもっと複雑になりそうですが、これで大きな JSON データでもストリーミング処理ができることがわかりました。一つ注意すべきは、ストリーミング処理の途中で JSON データをすべて受け取る前に回線が切れたりして中断する可能性があるということです。再試行は出来るかもしれませんが、その時に途中まで処理を行っていることになるので、場合によっては処理の取り消しなどが行えるように設計しておかなければならないかもしれません。