ストリームを複製し、両方の部分をストリーミング方式で処理するにはどうすればよいですか?

Question

時々、レポートやその他の二次的な用途のためにパイプラインに何かを挿入したいことがあります。 のように単純なものもあればwc -l、 のような複雑なものawkや、Python スクリプトのようなものもあります。次のようなパイプラインを実行できれば便利です。

zcat my_data_file.gz \
| wc -l > /tmp/linecount
| process_data.py

問題は、ほとんどのユーティリティがデータを stdout に書き込まないことです。teeデータを一時ファイルに書き込むことはできますが、すべてが完了するまで待たなければなりません。

zcat my_data_file.gz \
| tee /tmp/f \
| process_data.py && \
wc -l /tmp/f > /tmp/linecount && rm /tmp/f

これは最適ではありません。パイプラインの実行時間が非常に長くなる可能性があり、アナログからの中間結果をよりwc早く確認したい可能性があり、すべてのデータを一時ファイルに保存したくない可能性があります。

Answer 1

teeこれには置換を使用して処理できます>(...):

zcat my_data_file.gz |

# Count number of lines in stream
tee >(wc -l > /tmp/linecount) |

# Further processing
process_data.py

パイプは行の継続に使用でき、コマンド間にコメントを挿入できることに注意してください。これは、複雑なパイプラインを構築するときに便利な機能です。

Answer 2

完全に効率的ではありませんが、これを実現するには名前付きパイプ、これを使って作成できますmkififo(1)

質問の例では、

mkfifo /tmp/f

wc -l /tmp/f > /tmp/linecount &

zcat my_data_file.gz \
| tee /tmp/f \
| process_data.py &

wait

rm /tmp/f

&両方に とパイプラインが追加されていることに注意してくださいwc。これは、シェルがタスクをバックグラウンドにプッシュすることを意味します。 then の呼び出しは、waitすべてのバックグラウンド タスクが完了するまで待機します。両方のプロセスはほぼ同時に終了します。

プロセスの 1 つが大幅に遅い場合、そのプロセスteeの stdout パイプまたは書き込み先の名前付きパイプのいずれかがブロックされる可能性があるため、全体の速度が大幅に低下する可能性があることに注意してください。編集: また、二次プロセスが失敗した場合にパイプが破損して T 字型部分が終了するため、障害モードも増えました。