FilebeatとElasticsearch Ingest PipelineによるCSVファイル解析の実践ガイド

Elasticsearch 5の最も注目すべき新機能のひとつがingest nodeです。この機能により、Logstashスタイルの処理をElasticsearchクラスター内で実現でき、別途サービスやインフラを用意することなく、データをインデックス化する前に変換処理を行えます。以前、LogstashでCSVファイルを解析する方法についての記事を公開しましたが、本記事では比較のため、そのingest pipeline版をご紹介します。

ここでは、Filebeatを使ってデータをingest pipelineへ送信し、インデックス化したうえで、Kibanaで可視化するまでの一連の手順を解説します。

使用するデータ

無料で利用できる優れたデータソースは数多くありますが、ObjectRocketのメンバーの多くはテキサス州オースティン在住のため、今回はdata.austintexas.govのデータを使用します。レストラン検査データセットは適度なサイズであり、実務的な例題として十分な情報量を備えています。

以下は、このデータセットの一部を抜粋したもので、データ構造を確認できます。

Restaurant Name,Zip Code,Inspection Date,Score,Address,Facility ID,Process Description
Westminster Manor,78731,07/21/2015,96,"4100 JACKSON AVE
AUSTIN, TX 78731
(30.314499, -97.755166)",2800365,Routine Inspection
Wieland Elementary,78660,10/02/2014,100,"900 TUDOR HOUSE RD
AUSTIN, TX 78660
(30.422862, -97.640183)",10051637,Routine Inspection

おっと… 1エントリにつき1行という扱いやすい形式ではありませんね。しかし心配はいりません。ご覧のとおり、Filebeatには複数行エントリを処理し、データ中の改行に対処する組み込み機能があります。

編集者注：当初は「落とし穴」の少ないシンプルな例を想定していましたが、Elastic Stackが提供する各種ツールを駆使してこうしたケースに対処する過程を見せるのも興味深いと考え、あえてこのデータセットを採用しました。

Filebeatのセットアップ

最初のステップは、Filebeatを準備してElasticsearchクラスターへのデータ送信を開始できるようにすることです。Filebeatをダウンロード（ESクラスターと同じバージョンの利用を推奨）して展開したら、同梱のfilebeat.yml設定ファイルを使って非常に簡単にセットアップできます。今回のシナリオで使用した設定は以下のとおりです。

filebeat.prospectors:
- input_type: log
  paths:
    - /Path/To/logs/*.csv

  # 列見出しのある先頭行を除外
  exclude_lines: ["^Restaurant Name,"]

  # 最後の2列をエントリの終端として識別し、前の行を先頭に連結
  multiline.pattern: ',\d+,[^\"]+$'
  multiline.negate: true
  multiline.match: before

#================================ Outputs =====================================

output.elasticsearch:
  # 接続先ホストの配列。
  hosts: ["https://dfw-xxxxx-0.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-1.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-2.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-3.es.objectrocket.com:xxxxx"]
  pipeline: "inspectioncsvs"

  # プロトコルおよび基本認証の認証情報（任意）。
  username: "esuser"
  password: "supersecretpassword"

内容は非常にシンプルで、入力ファイルの場所と取得方法を指定するセクション、データの送信先を指定するセクションに分かれています。ここで特に触れておきたいのは、multiline設定とElasticsearch出力設定の2点です。

このデータセットはフォーマットが厳密ではなく、ダブルクォートの使い方が不統一で、改行も随所に混在しています。そこで最善のアプローチは「エントリの終端」を検出することでした。終端は「数値ID＋検査タイプ」という安定したパターンで構成され、ダブルクォートや改行を含みません。パターンに一致しない行はFilebeatによってキューイングされ、一致する最終行の前に連結されます。もしデータがきれいで、1エントリ＝1行の形式に従っているなら、multiline設定は基本的に不要です。

Elasticsearch出力セクションは標準的な設定に、pipeline:ディレクティブで使用するパイプライン名を追加しただけのものです。ObjectRocketサービスをご利用の場合は、管理UIの「Connect」タブから出力スニペットをコピーできます。正しいホスト情報が事前に入力されているため、pipeline行を追記し、ユーザー名とパスワードを記入するだけで完了です。また、まだの場合はシステムのIPアドレスをクラスターのACLに追加しておくことを忘れないでください。

Ingest Pipelineの作成

入力データとFilebeatの準備が整ったので、ingest pipelineを作成・調整しましょう。パイプラインが担う主なタスクは以下のとおりです。

• CSVの内容を正しいフィールドに分割する
• 検査スコアを整数型に変換する
• @timestampフィールドを設定する
• その他のデータフォーマットを整形する

以下は、これらすべてを担うパイプライン定義です。

PUT _ingest/pipeline/inspectioncsvs
{
  "description" : "Convert Restaurant inspections csv data to indexed data",
  "processors" : [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{REST_NAME:RestaurantName},%{REST_ZIP:ZipCode},%{MONTHNUM2:InspectionMonth}/%{MONTHDAY:InspectionDay}/%{YEAR:InspectionYear},%{NUMBER:Score},\"%{DATA:StreetAddress}\n%{DATA:City},?\\s+%{WORD:State}\\s*%{NUMBER:ZipCode2}\\s*\n\\(?%{DATA:Location}\\)?\",%{NUMBER:FacilityID},%{DATA:InspectionType}$"],
        "pattern_definitions": {
          "REST_NAME": "%{DATA}|%{QUOTEDSTRING}",
          "REST_ZIP": "%{QUOTEDSTRING}|%{NUMBER}"
        }
      }
    },
    {
      "grok": {
        "field": "ZipCode",
        "patterns": [".*%{ZIP:ZipCode}\"?$"],
        "pattern_definitions": {
          "ZIP": "\\d{5}"
        }
      }
    },
    {
      "convert": {
        "field" : "Score",
        "type": "integer"
      }
    },
    {
      "set": {
        "field" : "@timestamp",
        "value" : "//"
      }
    },
    {
      "date" : {
        "field" : "@timestamp",
        "formats" : ["yyyy/MM/dd"]
      }
    }
  ],
  "on_failure" : [
    {
      "set" : {
        "field" : "error",
        "value" : " - Error processing message - "
      }
    }
  ]
}

Logstashと異なり、ingest pipelineには（執筆時点で）CSV専用のプロセッサ／プラグインが存在しないため、変換処理は自前で実装する必要があります。今回は各行のカラム数が少なかったため、grokプロセッサで主要な処理を担わせました。カラム数が多いデータではgrokパターンが非常に複雑になりがちなので、その場合はsplitプロセッサとpainlessスクリプトを組み合わせ、反復的に処理する方法もあります。なお、2番目のgrokプロセッサは、このデータセットで郵便番号が2種類の書式で記録されていたことに対応させるために設けています。

デバッグ用途として、すべてのエラーを捕捉し、失敗したプロセッサの種類と原因となったメッセージを出力する汎用的なon_failureセクションも含めています。これにより、errorフィールドが設定されたドキュメントをインデックスから検索し、simulate APIでデバッグできるため、トラブルシューティングが格段に容易になります。詳細は後述します。

パイプラインのテスト

ingest pipelineの設定が完了したら、simulate APIでテストしてみましょう。まずサンプルドキュメントが必要です。方法はいくつかあります。pipeline設定を外してFilebeatを実行し、Elasticsearchから未処理のドキュメントを取得してもよいですし、コンソール出力を有効化してFilebeatを実行する方法もあります。後者は、Elasticsearchセクションをコメントアウトし、ymlファイルに以下を追記するだけです。

output.console:
  pretty: true

筆者の環境から取得したサンプルドキュメントがこちらです。

POST _ingest/pipeline/inspectioncsvs/_simulate
{
  "docs" : [
    {
      "_index": "inspections",
      "_type": "log",
      "_id": "AVpsUYR_du9kwoEnKsSA",
      "_score": 1,
      "_source": {
        "@timestamp": "2017-03-31T18:22:25.981Z",
        "beat": {
          "hostname": "systemx",
          "name": "RestReviews",
          "version": "5.1.1"
        },
        "input_type": "log",
        "message": "Wieland Elementary,78660,10/02/2014,100,\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\",10051637,Routine Inspection",
        "offset": 2109798,
        "source": "/Path/to/my/logs/Restaurant_Inspection_Scores.csv",
        "tags": [
          "debug",
          "reviews"
        ],
        "type": "log"
      }
    }
  ]
}

そしてレスポンスです（設定対象のフィールドのみに絞って表示しています）。

{
  "docs": [
    {
      "doc": {
        "_id": "AVpsUYR_du9kwoEnKsSA",
        "_type": "log",
        "_index": "inspections",
        "_source": {
          "InspectionType": "Routine Inspection",
          "ZipCode": "78660",
          "InspectionMonth": "10",
          "City": "AUSTIN",
          "message": "Wieland Elementary,78660,10/02/2014,100,\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\",10051637,Routine Inspection",
          "RestaurantName": "Wieland Elementary",
          "FacilityID": "10051637",
          "Score": 100,
          "StreetAddress": "900 TUDOR HOUSE RD",
          "State": "TX",
          "InspectionDay": "02",
          "InspectionYear": "2014",
          "ZipCode2": "78660",
          "Location": "30.422862, -97.640183"
        },
        "_ingest": {
          "timestamp": "2017-03-31T20:36:59.574+0000"
        }
      }
    }
  ]
}

パイプラインは確実に成功しており、何より重要なのはすべてのデータが意図したフィールドに正しく配置されている点です。

Filebeatの実行

Filebeatを実行する前に、最後にもうひとつだけ作業があります。ingest pipelineの挙動を確認するだけならこの手順は任意ですが、grokプロセッサで抽出したLocationフィールドをgeo_point型として活用したい場合は、filebeat.template.jsonファイルにマッピングを追加する必要があります。propertiesセクションに以下を追記してください。

"Location": {
  "type": "geo_point"
},

これで準備完了です。次のコマンドでFilebeatを起動しましょう。

./filebeat -e -c filebeat.yml -d "elasticsearch"

データの活用

まずはドキュメント件数を確認します。

GET /filebeat-*/_count
{}

{
  "count": 25081,
  "_shards": {
    "total": 5,
    "successful": 5,
    "failed": 0
  }
}

良好な結果です。次に、エラーが発生していないか確認してみましょう。

GET /filebeat-*/_search
{
    "query": {
        "exists" : { "field" : "error" }
    }
}

{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "failed": 0
  },
  "hits": {
    "total": 0,
    "max_score": null,
    "hits": []
  }
}

こちらも問題なし。エラーはゼロです。

これでKibanaでの可視化準備が整いました。ダッシュボード作成手順の詳細は別の機会に譲りますが、日付・レストラン名・スコア・位置情報が揃っているため、魅力的なビジュアライゼーションを作る自由度は十分にあります。

まとめ

繰り返しになりますが、ingest pipelineは非常に強力で、変換処理を手軽に実装できます。処理をすべてElasticsearch側に集約し、ホスト上には軽量なBeatsだけを置けば、パイプライン途中にLogstashを挟む必要はありません。一方で、ingest nodeにはLogstashと比べた場合のギャップも残されています。たとえば利用可能なプロセッサの種類がまだ限られており、CSVの解析といった一見単純なタスクでも、Logstashほど簡単には実現できません。とはいえ、Elasticsearchチームは新しいプロセッサを継続的にリリースしているため、両者の差は今後ますます縮まっていくことが期待できます。