Logstashを使ってCSVデータをElasticsearchにロードする方法

元記事は2015年9月10日、ObjectRocket.com/blogにて公開されました。

新しいElasticsearch®インスタンスを用意したものの、検索したい有用なデータがすべてCSVファイルの中にある――そんなことはありませんか?ご安心ください。Logstash®を使えば、ほぼあらゆる形式のデータを、Elasticsearchのインデックス内で簡単に検索できる形へと変換できます。

準備するもの

まずは対象となるデータと、本記事の例を実行できるUnix®系の環境が必要です。Windows®でも多少の調整を加えれば問題なく動作します。ここでは、Davis Vantage Pro2®気象ステーションからエクスポートした.CSV形式のデータを取り込み、新しいインデックスを作成することを目標にします。

ローカルファイルには、以下のような行が数百万行保存されているものとします。

$ head -3 /home/erik/weather.csv
HumOut,TempIn,DewPoint,HumIn,WindDir,RainMonth,WindSpeed,RainDay,BatteryVolts,WindChill,Pressure,time,TempOut,WindSpeed10Min,RainRate
76,78.0,78.227017302825,44,109,2.0,2,0.0,1.236328125,90.87261657090625,29.543,2015-06-18T17:49:29Z,86.5,1,0.0
76,78.0,78.227017302825,44,107,2.0,2,0.0,1.236328125,90.87261657090625,29.543,2015-06-18T17:49:45Z,86.5,1,0.0
76,78.0,78.32406784157725,44,107,2.0,0,0.0,1.236328125,90.83340000000001,29.543,2015-06-18T17:50:00Z,86.59999999999999,1,0.0

注意: この手順を実行するには、少なくとも1つのデータソースが必要です。

Logstashのセットアップ

データが用意できたら作業を始めましょう。まず、Javaがインストールされていることを確認します。

$ java -version
openjdk version "1.8.0_51"

JVM(Java Virtual Machine)であれば、OpenJDK®でもOracle®でもどれでも構いません。

続いて、Logstashをダウンロードして展開し、設定ファイル用のディレクトリを作成します。

$ curl -O https://download.elastic.co/logstash/logstash/logstash-1.5.4.tar.gz
$ tar xfz logstash-1.5.4.tar.gz
$ cd logstash-1.5.4
$ mkdir conf

設定ファイルの作成

inputセクション:データの読み込み元を定義する

まずはinputセクションを定義し、Logstashにデータの場所を伝えます。

input {
    file {
        path => "/home/erik/weather.csv"
        start_position => beginning

    }
}

これは単純に、Logstashに対してどこを見ればよいか、そしてファイルの先頭から読み込みたいことを指示しています。次に必要になるのがフィルターです。Logstashにはデフォルトで多数のフィルタープラグインが用意されており、この例ではそのうちのいくつかを使ってデータを解析します。現時点では、Logstashはファイル内のデータについて何も知らない状態なので、フォーマットや各フィールドの扱い方などを明示的に指定する必要があります。

filterセクション:データを解析する

filter {
    csv {
        columns => [
          "HumOut",
          "TempIn",
          "DewPoint",
          "HumIn",
          "WindDir",
          "RainMonth",
          "WindSpeed",
          "RainDay",
          "BatteryVolts",
          "WindChill",
          "Pressure",
          "time",
          "TempOut",
          "WindSpeed10Min",
          "RainRate"
        ]
        separator => ","
        remove_field => ["message"]
        }
    date {
        match => ["time", "ISO8601"]
    }
    mutate {
        convert => ["TempOut", "float"]
    }
}

カラム名を見れば内容はおよそ想像がつくと思いますが、それぞれの処理をもう少し詳しく説明します。

• まず、行全体の内容を含むmessageフィールドを削除しています。特定の属性を検索したいだけなので、このフィールドは不要だからです。
• 次に、timeフィールドがISO8601形式の日付であることを指定しています。これにより、Elasticsearchはそれをただの文字列ではなく日付として認識します。
• 最後に、mutate関数を使ってTempOutの値を浮動小数点数に変換しています。

outputセクション:Elasticsearchへ出力する

最後に、以下のコードでデータをElasticsearchに登録します。ホストとポート、認証情報、そして格納先インデックスの名前を設定してください。

output {
    elasticsearch {
        protocol => "https"
        host => ["iad1-20999-0.es.objectrocket.com:20999"]
        user => "erik"
        password => "mysupersecretpassword"
        action => "index"
        index => "eriks_weather_index"
    }
    stdout { }
}

Logstashの起動

設定が完了したら、いよいよ起動してみましょう。正常に動作していれば、出力は次のようになります。

$ bin/logstash -f conf/logstash.conf -v
Logstash startup completed

結果の確認

うまくいったでしょうか?Elasticsearchに問い合わせてみましょう。

$ curl -u erik:mysupersecretpassword 'https://iad1-20999-0.es.objectrocket.com:20999/_cat/indices?v'
health status index               pri rep docs.count store.size pri.store.size
green  open   eriks_weather_index 5   1   294854     95.8mb     48.5mb

ドキュメントが登録されていますね。続いて、実際に1件クエリしてみます。

$ curl -u erik:mysupersecretpassword 'https://iad1-20999-0.es.objectrocket.com:20999/eriks_weather_index/_search?q=TempOut:>75&pretty&terminate_after=1'

このクエリは、TempOutが75より大きいドキュメントを検索し(TempOut:>75)、人間が読みやすいように整形し(pretty)、シャードごとに最大1件までしか結果を返さないよう(terminate_after=1)指示しています。実行すると、次のようなレスポンスが返ってきます。

{
  "took" : 4,
  "timed_out" : false,
  "terminated_early" : true,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "failed" : 0
  },
  "hits" : {
     "total" : 5,
     "max_score" : 1.0,
       "hits" : [ {
    "_index" : "eriks_weather_index",
      "_type" : "logs",
      "_id" : "AU-yXZJIJb3HnhKvpdNC",
      "_score" : 1.0,
      "_source":{"@version":"1","@timestamp":"2015-06-22T10:24:23.000Z","host":"kibana","path":"/home/erik/weather.csv","HumOut":"86","TempIn":"79.7","DewPoint":"70.65179649787358","HumIn":"46","WindDir":"161","RainMonth":"2.7","WindSpeed":"0","RainDay":"0.36","BatteryVolts":"1.125","WindChill":"82.41464999999999","Pressure":"29.611","time":"2015-06-22T10:24:23Z","TempOut":75.1,"WindSpeed10Min":"0","RainRate":"0.0"}
    } ]
   } 
}

成功です!Logstashは、手元に眠るあらゆるデータをElasticsearchで自由に扱える形に変換してくれる頼れる「十徳ナイフ」のようなツールです。ぜひ活用してみてください。

まとめ

本記事では、CSV形式の気象データを例に、Logstashのinput・filter・outputという3つの基本設定を使って、Elasticsearchへのデータ取り込みを行う一連の手順を紹介しました。ポイントは以下の通りです。

• input:ファイルパスと読み込み開始位置を指定する
• filter:csvプラグインでカラムを分解し、dateプラグインで日付型を認識させ、mutateで数値変換を行う
• output:接続先・認証情報・インデックス名を指定してElasticsearchへ送信する

詳細やご質問については、www.rackspace.comにアクセスし、Sales Chatをご利用ください。Feedbackタブからコメントやお問い合わせをお寄せいただくことも可能です。