概要

大量のデータを処理するとき、メモリにすべてを展開せずにstream的に処理する必要があるときがある。
こういうstream的な処理は、大体どの言語でもあまりサンプルがなく、調べながらやるしかない。
ということでgoでやってみたのが以下。

Streamingの利点

メモリの使用が最小限であること
json linesフォーマットと親和性が高い

補足> json linesフォーマットとは？

jsonフォーマットはこんな感じ。

"list": [
  { "hoge": 1, "fuga": 2 },
  { "hoge": 1, "fuga": 2 },
  { "hoge": 1, "fuga": 2 },
  { "hoge": 1, "fuga": 2 },
  { "hoge": 1, "fuga": 2 }
]

json linesフォーマットはこんな感じ。

{ "hoge": 1, "fuga": 2 }
{ "hoge": 1, "fuga": 2 }
{ "hoge": 1, "fuga": 2 }
{ "hoge": 1, "fuga": 2 }
{ "hoge": 1, "fuga": 2 }

jsonファイルの拡張子は.jsonだが、json linesの拡張子は.jsonlだったり.jsonだったり固定ではない模様。

Streamingの欠点

通常のjsonフォーマットが使えない事
どうしてもgoroutineを使った非同期処理が必要となる事

なので、データ量が多くないなら、Streamingに処理せず普通にメモリに展開したほうが良いかもしれない。

Lambda

以下のサンプルは、Dynamodbにデータが挿入・更新されると実行される Dynamodb streamsという仕組みで動くLamba。
出力先のS3ファイルは、一行が一つのdynamodbレコードに相当するjsonであるような、json linesフォーマット。

func handler(ctx context.Context, event events.DynamoDBEvent) error {
	r, w := io.Pipe()

	go func() {
		defer func() {
			if err := w.Close(); err != nil {
				log.Println("writer close error: ", err)
				return
			}
		}()
		for _, r := range event.Records {
			if err := UnmarshalDynamoEvent(r.Change.NewImage, &data); err != nil {
				log.Println("unmarshal dynamo event error: ", err)
				w.CloseWithError(err)
				return
			}
			b, err := json.Marshal(&data)
			if err != nil {
				log.Println("json marshal error: ", err)
				w.CloseWithError(err)
				return
			}
			if _, err := w.Write(b); err != nil {
				log.Println("data buffer write error: ", err)
				w.CloseWithError(err)
				return
			}
			if _, err := w.Write(lf); err != nil {
				log.Println("LF buffer write error: ", err)
				w.CloseWithError(err)
				return
			}
		}
	}()

	key := fmt.Sprintf("%s/%s/%s.json", keyPrefix, now.Format("20060102"), now.Format("150405"))
	result, err := s3uploader.Upload(&s3manager.UploadInput{
		Body:   r,
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	})
 	log.Println("result:", result)
	if err != nil {
		return err
	}
	return nil
}

こんな感じでStreamingにS3に出力することができた。
ちなみにPipeは以下のようなもの。

io.Pipe

func Pipe() (*PipeReader, *PipeWriter){

io.PipeWriter

func (w *PipeWriter) Write(data []byte) (n int, err error) {
func (w *PipeWriter) Close() error {
func (w *PipeWriter) CloseWithError(err error) error {

データをWriterに書き込むわけだが、書き込み中になんらかのエラーが発生してStreamをクローズしたいとき、CloseWithErrorを使って失敗したことをReader側に通知できる。
成功したら普通にCloseで閉じる。

io.PipeReader

func (r *PipeReader) Read(data []byte) (n int, err error)
func (r *PipeReader) Close() error {
func (w *PipeReader) CloseWithError(err error) error {

Writer側で何等かのエラーが発生（CloseWithError）すると、Readに失敗してerrが帰ってくる。なので、エラーが帰ってきたら処理を中断する。
Read側でエラーが発生してWriter側に通知したい場合はこちらのCloseWithErrorを使うことになるが、main threadをRead側にするのかWrite側にするのかで通知の方向が変わりそう。

結論

io.Pipeを使う事で、データを書きこみながらio.Readerに流すことができる
Write中に処理を中断する必要がある場合、CloseWithErrorでReaderに通知する
json lines形式を入出力に使えば、streamingな処理を書きやすい
writerのCloseを忘れずに！（closeを忘れるとdeadlockになる）

goroutineを使うコードを書くのは抵抗があるが、こういうコードを書く場合逆にgoroutineを使ったほうが綺麗に書けるきもする。
golangと仲良くなりたいが、悩みはつきない・・・