2014年2月19日水曜日

Redshiftで実行中のqueryをkillする

以下のページにまとめられていました。
Cancel a query

'stv_recents'テーブルから実行中のquery(status='Running')を検索してkillしたいqueryのpidを取得し、cancelコマンドをpid指定して実行すればOKです。
select pid, trim(user_name), starttime, substring(query,1,20) 
from stv_recents
where status='Running';
cancel [pid];

投稿者 時刻: 0 件のコメント:

2014年2月7日金曜日

CPUのcoreごとの使用率を調べたい

メモ
$ mpstat -P ALL
10時19分43秒  CPU    %usr   %nice    %sys %iowait    %irq   %soft  %steal  %guest   %idle
10時19分43秒  all    8.73    0.01    1.20    0.19    0.00    0.10    0.04    0.00   89.74
10時19分43秒    0   17.30    0.00    2.28    0.44    0.00    0.32    0.06    0.00   79.60
10時19分43秒    1    7.72    0.01    1.02    0.12    0.00    0.04    0.04    0.00   91.06
10時19分43秒    2    5.24    0.01    0.80    0.11    0.00    0.02    0.03    0.00   93.79
10時19分43秒    3    4.69    0.01    0.70    0.09    0.00    0.01    0.04    0.00   94.46
４つcoreがある場合は上記のような形で表示されます。

定期実行する場合は、上記にプラスして引数に数字（例えば１）を指定すると、指定した秒数ごとに最新の情報が表示されます。
$ mpstat -P ALL 1

投稿者 時刻: 0 件のコメント:

2014年2月5日水曜日

ITエンジニア外出７つ小道具

インターネット時代のITエンジニアの宿命として、24時間365日、いかなる場所でも仕事をしなければいけない状況に陥る事があります。
そのため、多くの技術者もそうだと思いますが、常に持ち歩く道具というものが存在します。ここでは、ノートPCや携帯などの大きめな道具を除いた、僕が常に持ち歩く７つ小道具について紹介します。


１．WiFiルーター
外出先でもインターネットに接続するため。写真のものは、イーモバイルのPocketWiFiのLTE版です。
スマートフォンのテザリング機能を用いてネットワーク通信を行っても良いのですが、スマフォでは電源の持ちが心許ないので、電源を分散する意味でも専用のものを持ち歩いた方が便利です。WiFiルーターも、余計な機能がついてなく（ex.液晶タッチパネル）、バッテリーが大容量で、通信速度が速いものを選ぶのが望ましいです。
日本ではイーモバイルを使っていますが、外国に行くときはその国ごとにWiFiルーターをレンタルします。

２．iPhone/iPod向けLightningケーブル
iPhone持ち向け。急な充電や開発の際に使用。純正品が望ましいです。

３．MicroUSBケーブル
iPhone以外の多くのモバイル機器（Android端末や、上述のWiFiルーターなど）がMicroUSBに対応しているので。急な充電や開発の際に使用。純製のケーブルでないとAndroid端末が開発者モード時に認識してくれない事があるので、Android端末の純正品で揃えるのが望ましいです。

４．VGAアダプタ
急な会議やプレゼンなどで使用する事があるため。
HDMI対応のディスプレイやプロジェクタがもっと普及してくれればHDMIケーブルに置き換えます。

５．USB Ethernet アダプタ
主に最近のMacbook Pro/Airを持ち運ぶ人向け。有線LANしかインターネット環境が無い時に備えて。

６．SDカードリーダー
デジカメなどで撮影した写真を使用して資料やレポートなどを作成する事が必要な時があるため。多くのノートPCではSDカード読み取りスロットがありますが、様々なメディアからでも吸い出しできるように、一応。

７．モバイルバッテリー
写真には写ってないですが。電子機器の電池切れは死を意味します。


以前はこれ以外に以下のものも持ち歩いていました。

  • USBマイク（急なskypeミーティングに対応するため）
  • USBメモリ

上記のような小道具をいつも持ち歩いていれば、たいていのケースで対応できて焦ることは無いかな、と思います。


ちなみに大道具は以下で、それぞれ必要に応じて持ち運びます。

  • ノートPC
  • タブレット
  • スマフォ（Android＋iOS の２台）
  • それぞれの機器の充電器

投稿者 時刻: 0 件のコメント:

2014年1月29日水曜日

SparkでS3上のデータを使用する

http://spark.incubator.apache.org/docs/latest/ec2-scripts.html

SparkではローカルファイルやHDFS上のファイル以外に、S3上のファイルもデータとして使用することができます。
読み込む際に、SparkContextにAWSのACCESS_KEYとSECRET_KEYを認識させる必要がありますが、ネット上では色々情報が錯綜していてちょっと良くわかりませんでした。
（Hadoopクラスタのcore-site.xmlに書くとか、S3のURLに含ませるとか）

0.8.1のSparkContext.scala(core/src/main/scala/org/apache/spark/SparkContext.scala)のソースを見てみたら、以下のようになっていました。
  /** A default Hadoop Configuration for the Hadoop code (e.g. file systems) that we reuse. */
  val hadoopConfiguration = {
    val env = SparkEnv.get
    val conf = SparkHadoopUtil.get.newConfiguration()
    // Explicitly check for S3 environment variables
    if (System.getenv("AWS_ACCESS_KEY_ID") != null &&
        System.getenv("AWS_SECRET_ACCESS_KEY") != null) {
      conf.set("fs.s3.awsAccessKeyId", System.getenv("AWS_ACCESS_KEY_ID"))
      conf.set("fs.s3n.awsAccessKeyId", System.getenv("AWS_ACCESS_KEY_ID"))
      conf.set("fs.s3.awsSecretAccessKey", System.getenv("AWS_SECRET_ACCESS_KEY"))
      conf.set("fs.s3n.awsSecretAccessKey", System.getenv("AWS_SECRET_ACCESS_KEY"))
    }
    // Copy any "spark.hadoop.foo=bar" system properties into conf as "foo=bar"
    Utils.getSystemProperties.foreach { case (key, value) =>
      if (key.startsWith("spark.hadoop.")) {
        conf.set(key.substring("spark.hadoop.".length), value)
      }
    }
    val bufferSize = System.getProperty("spark.buffer.size", "65536")
    conf.set("io.file.buffer.size", bufferSize)
    conf
  }
要するに

  • OSの環境変数にAWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEYが指定されていたら、それぞれHadoop Configurationのfs.s3(n).awsAccessKeyId/fs.s3(n).awsSecretAccessKeyにマップされる
  • Java（Scala）のSystem Propertiesに指定されている設定はそのままConfigurationにマップされる。
という風に環境変数は扱われているようです。

一例として、OSの環境変数に指定してあげると、spark-shellやpysparkなどでもS3からのインタラクティブロードが可能になります。
$ export AWS_ACCESS_KEY_ID=xxxxxx
$ export AWS_SECRET_ACCESS_KEY=xxxxxx
$ ./spark-shell
(snip..)
scala> val data = sc.textFile("s3n://(bucket)/(path))
scala> data.count()
res0: Long = 1607157
みたいな感じで。

OSの環境変数として指定するもよし、プログラム中で閉じるもよし、この辺はケースバイケースだと思います。
投稿者 時刻: 0 件のコメント:

2014年1月28日火曜日

Spark + YARN + CDHの相性

http://spark.incubator.apache.org/
Sparkは、Hadoop上で動作するクラスターコンピューティングの基盤です。
という説明で正確かは自信ありません。。

生のHadoop＋HDFS構成でMapReduceを動作させると基本的にHDFSに対してシーケンシャルアクセスが発生しますが、たとえばMapReduce上でiterableな解析処理を実行しようとすると毎回HDFSをシーケンシャルに読み込むために性能劣化につながります。
そのためSparkではRDD(Resilient Distributed Dataset)というデータ機構を独自に用意し、一度読み込んだデータはメモリ上にキャッシュするなどの対策でiterableな解析処理も多少は最適化するようになっています。

この特性を活かして、MLlibのような機械学習ツールや、Bagelのようなグラフ解析ツールも用意されています。
http://spark.incubator.apache.org/docs/latest/mllib-guide.html
http://spark.incubator.apache.org/docs/latest/bagel-programming-guide.html

また、インタラクティブコンソール上でscalaやpythonを用いてduck typing的に処理が書けるのも特徴です。

Sparkは独自のCluster Managerを持ち、この上で分散処理をコントロールします。Cluster Managerはstandaloneで動作させることもできますし、MesosやYARNなどのResource Managerに任せることもできます。スペック的には。。



Sparkでは、YARNと連携する際、ダウンロードしてきたソース・ファイルを用いてYARNの環境向けにビルド(assembly)しなおす必要があります。
ドキュメントは以下。
http://spark.incubator.apache.org/docs/latest/running-on-yarn.html

基本的には、YARNの動作環境にあわせて環境変数(SPARK_HADOOP_VERSION)に適切な値を設定し、「sbt assembly」を実行します。
指定すべきバージョンは以下が参考になります。
http://spark.incubator.apache.org/docs/latest/hadoop-third-party-distributions.html

試しに、CDH4.5とSpark8.1を組み合わせてYARN連携をしようとしてみました。マニュアル通りSparkのインストールディレクトリで以下を実行します。
./sbt/sbt clean
SPARK_HADOOP_VERSION=2.0.0-cdh4.5.0 SPARK_YARN=true ./sbt/sbt assembly
ただし上記を実行すると、コンパイル時にエラーになります。
エラーメッセージは以下のような感じで
https://gist.github.com/moaikids/8663958

要約すると、org.apache.spark.deploy.yarn.YarnAllocationHandler.scala内で
import org.apache.hadoop.yarn.api.records.{AMResponse, ApplicationAttemptId}
というimport文があるのですが、このうち「AMResponse」インターフェースが存在しないため、コンパイルエラーになります。

Apache Hadoopの例えばr2.0.5-alphaには含まれていますが、
http://hadoop.apache.org/docs/r2.0.5-alpha/api/org/apache/hadoop/yarn/api/records/AMResponse.html
CDH4.5には含まれていない、
http://archive.cloudera.com/cdh4/cdh/4/hadoop-2.0.0-cdh4.5.0/api/org/apache/hadoop/yarn/api/records/
という状態です。

色々調べてみると、現在のSparkはApache Hadoop2.2でstableになったクラスを前提としているようで、CDH4.5（Hadoop2.0ベース）ではYARN経由での連携は出来ないようです。
詳細については以下のMailing Listのやりとりが詳しいです。
https://groups.google.com/forum/#!topic/spark-users/T1soH67C5M4

基本的には、CDH4.5でSpark8.1を動かす場合はstandaloneモードが無難ですし、どうしてもYARN連携したい場合は現在（2014/01/28現在）Beta1のCDH5を使って構築してみるか、のいずれかの選択肢になります。



投稿者 時刻: 0 件のコメント:

jmapを用いたJVM Heapの状態調査

メモ。
参考：http://www.javainthebox.net/laboratory/JavaSE6/managementtools/mngtools.html

jmapは、JVMのHeapメモリ内の情報をdumpするためのツールです。
http://docs.oracle.com/javase/jp/6/technotes/tools/share/jmap.html
引数にjavaのプロセスIDを渡してあげると、当該プロセスのメモリ情報をダンプすることができます。
jmap -dump:format=b,file=[file name] [pid]
上記の例では、バイナリ形式でheapのダンプ情報を出力します。
ダンプしたファイルは色々なツールで可視化が可能ですが、javaの標準コマンドであるjhatが一番楽だと思います。
jhat [file name] 
引数のファイルには、上述のjmapで出力したダンプファイルを指定します。
このように指定してあげると、デフォルトでは7000番のポートでサーバーが立ち上がり、ブラウザで接続すると以下のような分析画面が表示されます。

お察しの通り、上記はJettyの起動プロセスからjmapでダンプした結果を可視化したものです。
解析画面にはいくつか項目があり、そのうち画面下部の「Heap Histogram」を選択するとHeap上に存在するインスタンスの数とバイトサイズを確認することができます。


このように、jmapは手軽にJVMのHeapの状態を確認するには便利なのですが、すべてのデータをダンプすると、JVMのHeap Sizeによってはダンプファイルのサイズが非常に大きくなり、解析が難しくなります。
僕の例だと、数GBのダンプファイルをjhatで読み込ませようとしたら、30分くらい解析に時間がかかった結果最終的にOutOfMemoryでプロセスが死にました。。。

たとえばHeap Histogramだけを知りたい、という場合、以下のオプションを指定してあげるとjhatの上記の例で表示したのと同じようなデータが取得できます。
jmap -histo [pid] > [file name]
結果は以下

なお、以下のように「live」オプションを付与すると、生存中のオブジェクトについてのみ結果出力されます。 
jmap -histo:live [pid] > [file name]

投稿者 時刻: 0 件のコメント:

2013年12月10日火曜日

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で受け取り、表示をする、という感じになっているように見えます。


投稿者 時刻: 0 件のコメント:

2013年11月22日金曜日

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') という風に書くことで取得できます。


投稿者 時刻: 0 件のコメント:

2013年10月24日木曜日

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

投稿者 時刻: 0 件のコメント:

2013年10月3日木曜日

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する
投稿者 時刻: 0 件のコメント:

2013年9月14日土曜日

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;


投稿者 時刻: 0 件のコメント:

2013年9月7日土曜日

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（圧縮アルゴリズム）についてもっと調べてみた
投稿者 時刻: 0 件のコメント:

2013年9月4日水曜日

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

投稿者 時刻: 0 件のコメント:

2013年8月29日木曜日

fluent-plugin-jsonbucket

Fluentdのプラグインをいくつか作成しているのですが、多くが実証実験中（？）なので、rubygemsなどへの公開テスト的に簡単なものを書いてみようということで、掲題のものを10分くらいで作ってみました。

https://github.com/moaikids/fluent-plugin-jsonbucket

このoutput pluginは、Fluentd内で処理をしているjsonデータ(msgpackデータ）を、 key : {json} のような形で一つのキーにひとまとめにするプラグインです。

何をいっているのか自分でもよくわからなくなってきましたが、
{“a”:1, “b”:2, “c”:3 }
というjsonデータ(msgpackデータ）を
{“bucket”: {“a”:1, “b”:2, “c”:3 } }
という様な形式に変換するプラグインです。

「何に使うの？？」というのが正直な感想だと思いますが、用途は一般的では無いのでヒミツです。


投稿者 時刻: 0 件のコメント:

2013年8月21日水曜日

Elastic MapReduceの実行状況をグラフ化するmunin plugin

最近muninのプラグインを作ることが多くなりましたが、掲題のようにElastic MapReduceのプラグインを作成してみました。AWSのCloudWatchで用意されているグラフは非常に見辛いので何とかしたかったのが、作成の動機です。。。

https://github.com/moaikids/munin-emr

内部ではPythonのAWS SDK実装である"boto"を用いてjobflowsの値を取得し、stateごとに件数をグラフ化している感じです。


インストール方法は、上記のgitからpluginを取得し、munin-nodeがインストールされている環境に設置するだけです。ただしあらかじめPython(2.6 >= )がインストールされていることと、先述の"boto"ならびにtimezone関連のutil "pytz" のインストールが必要です。

pip install boto
pip install pytz

git clone https://github.com/moaikids/munin-emr.git
cp munin-fluentd/plugins/emr/* /usr/share/munin/plugins/
ln -s  /usr/share/munin/plugins/emr_jobflows  /etc/munin/plugins/emr_jobflows

また、AWSへの接続を行うため、AWSへアクセスするためのcredential情報をmunin-nodeの設定ファイルに記載する必要があります。

$ vim /etc/munin/plugin-conf.d/munin-node 
[emr_jobflows]
user root
env.access_key_id [ACCESS_KEY_ID]
env.access_secret_key [ACCESS_SECRET_KEY]
env.region [REGION]
env.time_range 10
一番最後の"time_range"では、何分前のジョブ情報を取得するかを指定します（単位：分）。
内部実装的には、describe_jobflows APIのcreated_afterの値に、「現在時刻 - time_range」の値を渡すようになっています。

当該プラグインをmunin環境に適用すると、以下のようなグラフが表示されるようになります。

投稿者 時刻: 0 件のコメント:

2013年8月19日月曜日

awsのpython sdkを使ってEMRのジョブステータスを取得する。

https://github.com/boto/boto
pythonのAWS SDKのひとつ"boto"を用いて、EMRのジョブステータスの取得を行ってみました。
import boto.emr
import datetime

access_key_id = "….."
access_secret_key = "….."
conn = boto.emr.connect_to_region("ap-northeast-1", aws_access_key_id=access_key_id,aws_secret_access_key=acces    s_secret_key)

created_after = datetime.datetime.today() - datetime.timedelta(hours=24)
print created_after
jobflows = conn.describe_jobflows(None, None, created_after, None )
for jobflow in jobflows:
    print jobflow.jobflowid
    print "  " + jobflow.name
    print "  " + jobflow.loguri
    print "  " + jobflow.state
    print
表示される結果は、先のRuby版の記事の内容とだいたい同じです。
http://itsneatlife.blogspot.jp/2013/08/awsruby-sdkemr.html

投稿者 時刻: 0 件のコメント:

2013年8月16日金曜日

awsのruby sdkを使ってEMRのジョブステータスを取得する。

↓な感じになる。
以下は直近１日以内のデータのみを取得するよう、'created_after'パラメータに値を指定してます。
#!/usr/bin/env ruby
require 'rubygems'
require 'aws-sdk'

emr=AWS::EMR.new
created_after = Time.at(Time.now().to_i - (24 * 60 * 60)).iso8601
p created_after
resp = emr.client.describe_job_flows({:created_after => created_after})

resp.data[:job_flows].map{ |job|
    p job[:job_flow_id]
    p "    " + job[:name]
    p "    " + job[:log_uri]
    p "    " + job[:execution_status_detail][:state]
    p
}
実行結果例 
"hogehoge"
"    hogejob.name"
"    s3n://hoge-log/emr/logs/"
"    COMPLETED"


投稿者 時刻: 0 件のコメント:

2013年8月12日月曜日

Linuxのdateコマンドでunixtimeと日付情報の変換

何気に知っていると便利でした。
# 現在時刻をunixtimeで表示
$ date +%s
1376285909
# 指定したunixtimeを日付情報で表示
$ date --date @1376285909
2013年  8月 12日 月曜日 14:38:29 JST

投稿者 時刻: 0 件のコメント:

2013年8月6日火曜日

redisで特定の条件にマッチするkeyを削除する

redisのコマンドラインとxargsを使うと楽ですね。
#全削除
redis-cli KEYS "*" | xargs redis-cli DEL
#前方一致
redis-cli KEYS "prefix*" | xargs redis-cli DEL
#後方一致
redis-cli KEYS "*suffix" | xargs redis-cli DEL

投稿者 時刻: 0 件のコメント:

2013年8月2日金曜日

aws-sdk-rubyを使用してspot instanceのpricing historyを取得

https://github.com/aws/aws-sdk-ruby 
gem install aws-sdk

aws-sdk-rubyをインストールした上で、使用しているAWSのID情報を設定し 
export AWS_ACCESS_KEY_ID='...'
export AWS_SECRET_ACCESS_KEY='...'
export AWS_REGION='ap-northeast-1'
以下のようなruby srouceを実行すると、spot instanceのpricing historyが取得出来ます。
#!/usr/bin/env ruby
require 'rubygems'
require 'aws-sdk'

ec2=AWS::EC2.new
resp = ec2.client.describe_spot_price_history(
    :instance_types => ['m1.small', 'm1.medium', 'm1.large', 'm1.xlarge'],
    :start_time => (Time.now - 60 * 60 * 24).iso8601, 
    :availability_zone => 'ap-northeast-1b')

resp.data[:spot_price_history_set].map{ |history| 
    if history[:product_description] == "Linux/UNIX"
        p history[:instance_type] + " => " + history[:spot_price] + " (" + history[:timestamp].to_s + ")" 
    end
}
実行結果。
"m1.small => 0.017000 (2013-08-01 22:52:34 UTC)"
"m1.large => 0.067000 (2013-08-01 22:48:47 UTC)"
"m1.medium => 0.035000 (2013-08-01 22:48:44 UTC)"
"m1.xlarge => 0.134000 (2013-08-01 22:48:43 UTC)"
"m1.small => 0.000100 (2013-08-01 22:47:34 UTC)"
"m1.xlarge => 0.134000 (2013-08-01 13:12:40 UTC)"
"m1.medium => 0.035000 (2013-08-01 06:47:28 UTC)"
"m1.small => 0.017000 (2013-08-01 02:42:36 UTC)"
"m1.large => 0.067000 (2013-08-01 00:04:50 UTC)"
"m1.xlarge => 0.134000 (2013-07-31 13:11:39 UTC)"
"m1.medium => 0.035000 (2013-07-31 06:44:00 UTC)"


投稿者 時刻: 0 件のコメント:
登録: 投稿 (Atom)