Modern Linux Pipeline Programmingを補助してくれるツールたち

# Line-delimited JSON

NDJSON, LDJ, JDJSON, JSONLなどとも呼ばれることがある、1行1JSONのテキストことです。Dockerのjson-logや、jqがデフォルトで受け付ける入力としても有名ですね。

この形式のよいところは、対応しているツールが多いために、フォーマット変換を自前でやらなくても本質的な処理だけをパイプで繋げられる点です。

パイプラインの一部となるツールを自分で実装する場合は、入力と出力にこの形式を採用できないか検討すると、ツールの再利用性が高まります。

Ref: https://en.wikipedia.org/wiki/JSON_streaming#Line-delimited_JSON

## bethos

https://github.com/Jeffail/benthos

様々なプロトコルから/へのIN/OUTをつないで、Fan-in/outなど様々なパターンでルーティングすることができるstream multiplexer serviceと呼ばれるアプリケーション。

- Amazon (S3, SQS)
- File
- HTTP(S)
- Kafka
- MQTT
- Nanomsg
- NATS
- NATS Streaming
- NSQ
- RabbitMQ (AMQP 0.91)
- Redis
- Stdin/Stdout
- ZMQ4

入出力にそれぞれ標準入力と標準出力を指定できるので、bash pipelineの一部として利用することもできる。

標準入力を利用する場合、１行１メッセージのセマンティクス以外に、１行１パート・空行がメッセージの区切り・・・というマルチパートのセマンティクスも利用できて便利。

https://github.com/Jeffail/benthos/tree/master/resources/docs/inputs#stdin

```
sourceapp | benthos -c ./config.yaml
```

## mirocule

https://github.com/stackvana/microcule

> SDK and CLI for spawning streaming stateless HTTP microservices in multiple programming languages

２０以上のプログラミング言語で書かれたスニペットをHTTPサービスにラップするツール、とでも言えばよいだろうか。

https://github.com/stackvana/microcule#cli-examples

実は標準入力からリクエストを受けることができるので、出力を標準出力にしておけばPipelineの一部としても利用できる。

```console
tail -f ReadMe.md | microcule --stream=true ./examples/services/streams/echo.js
```

Linux Pipeline Programmingの文脈では、

- Webアプリケーションに仕立て上げたPipelineをHTTPでつなぐことで、PipelineをステートレスなWebアプリケーションとして水平スケールさせられる

という点が面白い。

## hook.io

https://github.com/bigcompany/hook.io

miroculeベースのWebhook as a Serviceのためのツール。

Linux Pipeline Programmingの文脈だと、

ベースとなるMiroculeが与える特性に加えて

- Webアプリケーションに仕立て上げたPipelineをHTTPでつなぐことで、PipelineをステートレスなWebアプリケーションとして水平スケールさせられる
- その管理ツール・コンソールを提供してくれる

という点が面白い。

## esbulk

https://github.com/miku/esbulk

１行１ JSON形式の入力をElasticsearchにバルクロード(index)してくれるツール。パイプラインの終端におけば、入力を変換したのちESに効率的にロードするパイプラインが組める。

```console
cat docs.ndjson | esbulk -index example file.ldj
```

## bashful

https://github.com/wagoodman/bashful

Bashコマンドを並列に、または選択的に実行し、実行経過・結果をヒューマンフレンドリーな形式でリアルタイム表示してくれるツール。

Linux Pipeline Programmingの文脈では、複数のPipelineから構成されるDurableでないWorkflowを、人間が実行する場合の良いインターフェースになる。

![demo.gif](https://raw.githubusercontent.com/wagoodman/bashful/master/demo.gif)

ref: https://github.com/wagoodman/bashful

## 　小ネタ

### inotify...

(いまさらだが)ローカルファイルの変更検知をこの文脈で気軽にやる方法はいまだにinotifyなので、覚えておいて損はない

参考: https://superuser.com/questions/181517/how-to-execute-a-command-whenever-a-file-changes

### bashでcommand-line flags...

(こちらもいまさらだが)bashスクリプト２引数以上とるようになってきたら２つ目以降はオプションとみなせるケースも多い。
そういう場合はコマンドラインフラグを活用する。

参考: https://stackoverflow.com/questions/192249/how-do-i-parse-command-line-arguments-in-bash

### seq & gnuparallel & xargs ...

（いまさらだが）bashスクリプト内で気軽にmap/reduceスタイルの並列処理を行いたい場合、この３つの組み合わせはイディオムとして覚えておくと便利。

例えば、以下の例では1-10番の入力を並列に処理して、結果を番号順に連結する。

```
seq 1 10 | parallel './dosomething foo/bar/{} > intermediate-{}.json'
seq 1 10 | xargs -I{} cat intermediae-{}.json > result.ndjson
```

参考: https://unix.stackexchange.com/questions/121967/how-to-run-parallel-processes-and-combine-outputs-when-both-finished

## 未来

### pachyderm

Kubernetes上で「ファイルを読み書きする任意のLinuxプログラム」を分散実行させてパイプラインをつくることができる、Pachydermというプラットフォームがある。

PachydermのIssueでは、以下のようなFeature Requestがでている

- [STDIN/OUTに読み書きするアプリケーションを実行単位としたい](https://github.com/pachyderm/pachyderm/issues/1851)
- [入力をバッチ処理したい](https://github.com/pachyderm/pachyderm/issues/1662)

これらができるようになると、Linux Pipelineをコンテナ化して分散実行することができ、さらにLinux Pipelineが気軽に、高速に大量データのストリーミング処理に応用できる範囲が増えると思われる。

### Argo

Kubernetes上で「コンテナ化されたアプリケーションのワークフロー」を実現するための、Argoというシステムがある。

Pipelineを「ファイルを読んでSTDIN経由でパイプラインに入力し、パイプラインの出力をファイルに書き出すようなコンテナ」でラップしてやれば、一応Linxu Pipelineから構成されるWorkflowをKubernetes上で実行することができる。

ただし、仕組み上、Argo単体で低レイテンシーなストリーミング処理はできないので、多くのケースでLinux Pipelineに求められると思われる「低レイテンシーなストリーミング処理」というユースケースには合わなさそうではある。