fluent-plugin-redshiftとその他pluginを組み合わせてredshiftへデータ保存

http://aws.amazon.com/jp/redshift/
AWSから安価で使用可能なDWH製品Redshiftが公開されてしばらく立ちます。
非常に興味深いサービスなのですが、Redshiftへのデータの登録が独特(S3に置いたCSV/TSVをcopyコマンドを用いて登録）という事もあり、ちょっと面倒くさく感じていました。

最近、掲題のようにredshiftへのデータ保存が行えるFluentdプラグインがあるのを発見したので、こちらと他プラグインを組み合わせて、Fluentdを用いたRedshiftへのデータ保存を試してみました。


◯fluent-plugin-redshift
https://github.com/hapyrus/fluent-plugin-redshift
BufferedOutputプラグインの一つで、仕組みとしてはchunk単位でS3にデータを書き込んだ上でcopyコマンドを使用してRedshiftにデータ登録を行っています。S3上のファイルはRedshift登録後もそのまま残るので、作業の証跡としてや、他用途への活用（EMRを使っての解析等）にも使用できるようになっています。

Redshftへの書き込みはchunk単位で行われるので、書き込み頻度については"buffer_chunk_limit"や"flush_interval"などの設定で調整します。

一つおもしろい形になっているのがデータ登録形式で、json形式のデータを投入したい場合

{"log":{"a":1."b":2,"c":3}}

といった形で、"log"というキーの中にjsonデータをすべて包含させる必要があります。
※"log"の値は設定ファイル中で"record_log_tag"に任意の値を指定することで変更可能です。

なぜこのような仕様になっているかは定かではないですが、現状実装がそのようになっていますし、TSV/CSV形式で流しこむのに比してスキーマ情報（カラム名）を明示的に指定できるメリットもあるので、JSONデータ形式で流し込みたいところです。
とはいえ、情報ソース側で出力するJSONの形式をredshiftプラグインの仕様にあわせて変更する、というのも手間です。


◯fluent-plugin-jsonbucket
こちらの課題を解決することだけを目的に、fluent-plugin-jsonbucketというプラグインを作りました。
https://github.com/moaikids/fluent-plugin-jsonbucket

こちらは、READMEに記載している通リ

{“a”:1, “b”:2, “c”:3 } -> {“bucket”: {“a”:1, “b”:2, “c”:3 } }

という形で、JSONデータを特定のkeyのvalue値として内包させるためのプラグインです。

jsonbucketと組み合わせる事で、情報ソース側で特別な加工をすることなく、pluginでJSONの変換処理を吸収させる事ができます。


◯fluent-plugin-reassemble
基本的には上記の組み合わせでOKですが、情報ソースのうちすべての項目をredshiftに保存するのも無駄が多いですし、redshiftに登録する時にデータの簡単な変換を行いたいケースもあります。
たとえば、JSON上は文字列として扱われている数値をredshift側ではintegerで保存したい、日付情報についてログ上ではunixtimeで保存されているけれど、見通しを良くしたりもしくはテーブル設計上の都合でredshift側では"yyyy-mm-dd"のdate型で保存したい、といったケースです。

また、元々のデータがネスト構造になっていたりするとredshift的には扱いづらいので、redshift保存のためにフラットな構成に置き換える必要があります。

その辺を情報ソース側ですべてコントロールするのもコレまた手間ですので、plugin側でこの手のtransform処理を行うfluent-plugin-reassembleを作成しました。
https://github.com/moaikids/fluent-plugin-reassemble

保存したいkeyの情報と、valueをどのように変換するかを設定ファイル上で記載することで、指定通りにデータの加工(reassemble)が行えるプラグインです。

こちらもREADMEに例を載せていますが、

ex1. 
assemble foo1:bar1:to_i, foo2:bar2:to_s
record => {"foo1": "1", "foo2": 2}
reassemble => {"bar1": 1, "bar2": "2"}

ex2. 
assemble foo1,foo2:bar2,foo3:bar3:unixtime_to_datetime,foo4:bar4:url_to_host
record => {"foo1": "1", "foo2": 2, "foo3": 1377946931, "foo4": "http://www.sada.co.jp/concert.html"}
reassemble => {"foo1": "1", "bar2": 2, "bar3": "2013-08-31 20:00:02", "bar4": "www.sada.co.jp"}

といった形で、"assemble"に所定のフォーマットで対象のキーとoperation内容を記述することで、plugin上でデータの加工が行えます。
(指定のフォーマットは以下)

{extract_key1}:{replaced_key1}:{operation1},{extract_key2}:{replaced_key2}:{operation2},....{extract_keyN}:{replaced_keyN}:{operationN}

こちらを使う事で、情報ソースのうち必要なデータを必要な形式で変換・加工し、redshiftに登録する、といった事がpluginだけで完結させる事ができます。
(もちろんreassembleプラグインについてはredshiftだけに限らず、すべての用途向けに仕えるpluginですが)


という感じで、これらのpluginを組み合わせる事で、非常に手軽にredshiftへデータの保存が行えるようになります。