Tableau上でRを使用する（インストール編）

Tableau Desktop 8.1からR scriptの実行がサポートされたようです。
http://www.tableausoftware.com/ja-jp/about/blog/2013/10/tableau-81-and-r-25327

Tableau上でRを実行するには、同じPC上（もしくはネットワーク接続可能な他のPC上）でRserveが動作している必要があります。
上記URLにも記載されていますが、セットアップ手順としては以下になります。

1. Rをインストール
2. Rを起動し、Rserveパッケージをインストール（ install.packages("Rserve"); ）
3. Rserveを起動（ library(Rserve); Rserve() ）
4. TableauからRserveに接続（ ヘルプ→設定とパフォーマンス→R接続の管理 ）

ただ、まっさらなR環境だと上記では足らず、追加でplyrパッケージのインストールが必要になります。（ install.packages("plry"); ）

これで、無事動作するようになります。

詳しくは追ってないですが、Tableau x Rのインテグレーションは、Tableauで収集したデータセットをRserve経由でR環境にわたし、実行結果を再度Tableauで受け取り、表示をする、という感じになっているように見えます。

RedshiftのJSON functions

単純に見落としていただけで以前から有ったのかもしれませんが、RedshiftにもJSON関連のfunctionが存在していました。。
http://docs.aws.amazon.com/redshift/latest/dg/json-functions.html

PostgreSQLにもJSON functionは各種存在しますが、Redshiftのそれについては関連性はなく、独自の実装に見受けられます。


◯JSON_ARRAY_LENGTH function
http://docs.aws.amazon.com/redshift/latest/dg/JSON_ARRAY_LENGTH.html
対象となる文字列がJSONのarray表現であるとして、そのarrayの長さを返却するfunctionです。

hoge=# select json, json_array_length(json) from hoge;
| 0
["a<","b","c"] | 3 

渡された文字列が空文字の場合、0が返却されます。
また、文字列以外の値など、JSONのparsingが行えない値を渡した場合はエラーになります。

ERROR:  JSON parsing error
DETAIL:  
  -----------------------------------------------
  error:  JSON parsing error
  code:      8001
  context:   invalid json array object 2013-11-21 00:36:08
  query:     2416921
  location:  funcs_json.h:81
  process:   query1_20 [pid=9891]
  -----------------------------------------------

◯JSON_EXTRACT_ARRAY_ELEMENT_TEXT function
http://docs.aws.amazon.com/redshift/latest/dg/JSON_EXTRACT_ARRAY_ELEMENT_TEXT.html
渡されたJSON Array文字列の中から指定した要素の値を取得します。

hoge=# select json, json_extract_array_element_text(json, 0);
 |  
["a","b","c"] | a

当該要素が存在しない場合は空文字が返ります。


◯JSON_EXTRACT_PATH_TEXT function
http://docs.aws.amazon.com/redshift/latest/dg/JSON_EXTRACT_PATH_TEXT.html
渡されたJSON文字列の中から、指定したkeyに合致するvalue値を取得します。

hoge=# select json, json_extract_path_text(json, 'a')
 | 
{"a":1, "b":2, "c": 3} | 1

該当するkeyが存在しない場合は空文字が返ります。
また、{"a": {"b":{"c" :3} } } のようにネスト構造のjsonから"a.b.c"のvalueを取得したい場合は、 json_extract_path_text(json, 'a', 'b', 'c') という風に書くことで取得できます。

Redshiftでの"String contains invalid or unsupported UTF8 codepoints"問題とACCEPTINVCHARSオプション

Redshiftにおけるvarcharなど文字列型は基本的にUTF-8文字列しかサポートしていないようなので、非UTF-8と思われる文字列が含まれているTSV/USVをRedshiftにロードしようとすると以下のエラーが発生します。

String contains invalid or unsupported UTF8 codepoints. Bad UTF8 hex sequence:

基本的には取り込むデータ中に非UTF-8と思われる文字コードが含まれないよう、前処理にてデータの整形を行うことが正攻法だと思います。
しかし試験的にデータを取り込みたかったり、ひとまずRedshiftにデータを入れたい場合は、'ACCEPTINVCHARS'オプションを用いて当該文字を他の文字にreplaceしてRedshiftに取り込む、ということができます。

http://docs.aws.amazon.com/redshift/latest/dg/r_COPY.html#acceptinvchars

ACCEPTINVCHARS AS '_'

のような形でオプション指定します。ASで置換後の任意の文字列を指定できます（デフォルトは'?')

'ACCEPTINVCHARS'を指定した場合、たとえ非UTF-8文字列が含まれていても自動的に置換されRedshiftに取り込まれます。なお、取り込み時に以下のようなメッセージが表示されます。

INFO: Load into table 'hoge' completed, 59 record(s) were loaded with replacements made for ACCEPTINVCHARS. COPY

参考：Amazon Redshift / Analytics / Knowledge Base - Papertrail Support

nginxのmerge_slashes

nginxや、その他多くのWebサーバーでは、リクエストされたURLの中の連続しているスラッシュをマージして解釈するような動きをする事が多いです。

たとえば

http://www.sada.co.jp//////index.html → http://www.sada.co.jp/index.html

のような感じで。

多くの場合上記の動作をしてくれて問題無いのですが、連続したスラッシュに意味がある場合もあり、その際は自動的にmerge slashされると都合が悪いケースもあります。

nginxの場合は"merge_slashes"というsyntaxがあるようで、こちらはデフォルトではonなのですが、offにすることで自動的にmerge slashをしないようにすることができるようです

merge_slashes off

こんな感じ。

参考：nginxはデフォルトでmerge slashする

redshiftのSQLでunixtimeをtimestamp型に変換

timestamp から unixtimeに変換するのは Extract 関数があるので楽ですが、逆の変換はto_timestampなどがredshiftに存在しないため面倒です。

以下のようなSQLを書くことで実現できます。なお、以下の場合は、カラムtime_stampに秒単位でunixtimeが入っている場合です。

select  TIMESTAMP 'epoch' + unix_time * INTERVAL '1 second' as time_stamp from hoge limit 10;

timezoneの変換を合わせて行う場合は、 CONVERT_TIMEZONE 関数を使用します。以下はtimezoneをJSTに変換する例。

select CONVERT_TIMEZONE('JST', TIMESTAMP 'epoch' + unix_time * INTERVAL '1 second') time_stamp from hoge limit 10;

redshiftのカラムに適切なencodeを設定する

redshiftはいわゆるcolumner型のデータ構造を持っているので、各columnに適切なencode(圧縮)方式を適用することが運用に際して非常に重要になります。


◯redshiftに任せる
redshiftは、当該テーブルへの初回copy時に適切なencodeを自動検出し適用する機能があります（10万レコードからサンプリングし、推定）。最もナイーブな方式としてはこちらに任せる、という方法があります。

適用されたencodeは、以下のクエリーを実行することで確認できます。

select * from pg_table_def where tablename = 'hoge';

◯analyze compressionを実行した結果を適用する
ただし、初回のサンプリングだけでは限界があり、運用を重ねていった結果実情に則さなくなるケースも多いです。

現在蓄えられたデータから最適なencodeを推薦する"analyze compression"というコマンドがredshiftには提供されています。
http://docs.aws.amazon.com/redshift/latest/dg/r_ANALYZE_COMPRESSION.html

analyze compression hoge;

こちらの実行結果と、初回copy時に割り当てられたencodeでは、同じ場合もありますし、異なる場合もあります.
異なる場合はどうするか？
alter tableなどで後付でencodeが変更できればよいですが、流石にそのような機能はredshiftに提供されていないので、基本的には以下の手順でtableを作り直します。

1. 既存のテーブルhogeと全く同じcolumnを持つtemporary table を作成。
2. hogeのデータをtemporary tableに退避(insert into hoge_tmp (select * from hoge))
3. hogeをdrop(drop table hoge)
4. 適切なencodeを指定して、hoge table を作りなおす(create table ~ )
5. temporary tableのデータをhogeにコピーする( insert into hoge (select * from hoge_tmp))
6. temproary table を削除

◯encodeの種類
以下にまとめられています。
http://docs.aws.amazon.com/redshift/latest/dg/c_Compression_encodings.html

run-lengthは、同一カラム内で同じ値を持つデータが連続して出現する可能性が高い場合、非常に有効です。

同じ値ではないけれど近しい値が連続して出現する場合は、差分情報を用いるdeltaならびにdelta32kが有効です。

数値系の値で、その型が持つサイズ(integer であれば32bit、bigintであれば64bit）を使いきらず、本来のサイズよりも少ないbit数で表現可能な値がカラム中に多い場合はmostly8/16/32が有効です。

辞書式圧縮が有効なケースの場合は、文字列の場合はtext255/32k、そうでない場合はbytedictになります。

いずれにも当てはまらず、圧縮が意味が無いケースの場合はraw（圧縮しないでそのまま）、という感じです。


◯脳筋的な運用をするのであれば定期的にanalyze compression → 作り直し
redshiftを信じるのであれば、何も考えずにanalyze compressionを定期的に実行し、必要であればその結果を元に作り変えるのが良いのでは無いかと思います。



なお、redshiftは、そのデータ構造の性質上、データの出現順序(insert順序）によって圧縮効率が大きく変わるようです。
圧縮しやすい形でデータを保存したいのであれば、ある程度データをあらかじめソートした上で投入するのが、効率的と思われます。
また、encodeの選定についても、どのような順序でデータが投入されるかをあらかじめ想定しておくと、最初のcreate tableの段階で妥当なencodeが何なのか類推しやすくなると思います。


※参考文献：
本記事と直接的な関連は無いですが、encodeの理解に非常に参考になったスライド
Redshift Compression Encodings（圧縮アルゴリズム）についてもっと調べてみた

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へデータの保存が行えるようになります。