はじめに

Embulk + Elasticsearch + kibana + OpenStreetMapでお手軽に地理情報を可視化

上の投稿では「Embulk + Elasticsearch + kibana」の組み合わせで、AEDが設置されている場所を簡単に地図へマッピングしており興味深い内容でした。
最終的に以下のようにkibanaの地図（OpenStreetMap）上にAEDをプロットして表示しています。

「Embulk + Elasticsearch + kibana」に興味があったので、投稿の内容を自分の環境でも試してみました。元記事の内容を試しただけなのですが、OS（元記事はMacOS）とElasticsearch/kibanaのバージョンが違うのでメモとして残しておきます。

環境

使用した環境は以下のとおりです。

CentOS 7.5
Elasticsearch 6.8.3
kibana 6.8.3

CentOSはfirewalldをあらかじめ無効に設定しています。
また、Elasticsearchとkibanaは以下からダウンロードします。

https://www.elastic.co/jp/downloads/past-releases/elasticsearch-6-8-3
https://www.elastic.co/jp/downloads/past-releases/kibana-6-8-3

最初は7.4を使用していたのですが、typeがなくなるなど大きく変更されており情報も少なかったため、6系の最新版である6.8.3を使用することにしました。

JDK8インストール

Elasticsearch/kibanaのインストール前に以下の手順でOpenJDK8をインストールしておきます。

# yum install java-1.8.0-openjdk-devel

# java -version
openjdk version "1.8.0_222"
OpenJDK Runtime Environment (build 1.8.0_222-b10)
OpenJDK 64-Bit Server VM (build 25.222-b10, mixed mode)

JAVA_HOMEを設定し、PATHにJavaのパスを追加しています。

# echo "export JAVA_HOME=$(readlink -e $(which java)|sed 's:/bin/java::')" > /etc/profile.d/java.sh
# echo "PATH=\$PATH:\$JAVA_HOME/bin" >> /etc/profile.d/java.sh
# source /etc/profile

AED情報入手

地図にプロットするのはAED情報を使用します。
AEDオープンデータプラットフォームから、東京都のAED情報をダウンロードします。

$ curl https://aed.azure-mobile.net/api/aedinfo/%E6%9D%B1%E4%BA%AC%E9%83%BD/ -o aedinfo_tokyo.json

Elasticsearchのインストール

以下の手順でElasticsearchをダウンロードし、インストールとサービスの自動起動まで設定します。

# wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.8.3.rpm
# rpm -ivh elasticsearch-6.8.3.rpm
# systemctl enable elasticsearch.service
# systemctl start elasticsearch.service

curlコマンドでElasticsearchにアクセスできることを確認します。

# curl http://localhost:9200
{
  "name" : "s0nDSnH",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "OFFAczRGRUKAqFcqyBnFsg",
  "version" : {
    "number" : "6.8.3",
    "build_flavor" : "default",
    "build_type" : "rpm",
    "build_hash" : "0c48c0e",
    "build_date" : "2019-08-29T19:05:24.312154Z",
    "build_snapshot" : false,
    "lucene_version" : "7.7.0",
    "minimum_wire_compatibility_version" : "5.6.0",
    "minimum_index_compatibility_version" : "5.0.0"
  },
  "tagline" : "You Know, for Search"
}

外部からも接続できるように設定ファイルを修正します。

# vi /etc/elasticsearch/elasticsearch.yml

#network.host: 192.168.0.1
network.host: 0.0.0.0 # アクセスできるIPアドレスを指定する。0.0.0.0で全て許可。
transport.host: localhost # 追加

設定変更後、Elasticsearchを再起動します。

# systemctl restart elasticsearch

kibanaのインストール

以下の手順でkibanaをダウンロードし、インストールとサービスの自動起動まで設定します。

# wget https://artifacts.elastic.co/downloads/kibana/kibana-6.8.3-x86_64.rpm
# rpm -ivh kibana-6.8.3-x86_64.rpm

# systemctl enable kibana.service
# systemctl start kibana.service

curlコマンドでkibanaにアクセスできることを確認します。

# curl http://localhost:5601

外部からも接続できるように設定ファイルを修正します。

# vi /etc/kibana/kibana.yml

#server.host: "localhost"
server.host: "0.0.0.0"

設定変更後、kibanaを再起動します。

# systemctl restart kibana.service

以下のアドレスでブラウザからアクセスすると、kibanaのトップ画面が開きます。

http://[IPアドレス]:5601

Embulkのインストール

公式サイトのQuick Startを参考にインストールします。

Quick Start

# curl --create-dirs -o ~/.embulk/bin/embulk -L "https://dl.embulk.org/embulk-latest.jar"
# chmod +x ~/.embulk/bin/embulk
# echo 'export PATH="$HOME/.embulk/bin:$PATH"' >> ~/.bashrc
# source ~/.bashrc

プラグインもインストールします。

# embulk gem install embulk-parser-jsonpath embulk-output-elasticsearch
# embulk gem install embulk-filter-insert embulk-filter-ruby_proc embulk-filter-column

Elasticsearchのスキーマ作成

schema.jsonを以下のように作成。

schema.json

{
  "index_patterns": ["aedmap*"],
  "settings": {
    "number_of_shards": 1
  },
  "mappings": {
    "aed_schema": {
      "properties": {
        "LocationName": {
          "type": "text"
        },
        "address": {
          "type": "text"
        },
        "location": {
          "type": "geo_point"
        }
      }
    }
  }
}

先ほど作成したスキーマファイルをElasticsearchに登録します。

# curl -H 'Content-Type:application/json' -XPUT localhost:9200/_template/aedmap -d @schema.json
{"acknowledged":true}

EmbulkでAEDファイルをElasticsearchへロードする

EmbulkでAEDファイルをElasticsearchへロードする設定ファイル（config.yml）を以下のように作成します。

config.yml

in:
  type: file
  path_prefix: ./aedinfo_tokyo.json
  parser:
    charset: UTF-8
    newline: CR+LF
    type: jsonpath
    columns:
    - {name: Id, type: long}
    - {name: LocationName, type: string}
    - {name: Perfecture, type: string}
    - {name: City, type: string}
    - {name: AddressArea, type: string}
    - {name: Latitude, type: double}
    - {name: Longitude, type: double}

filters:
  - type: insert
    columns:
      - location: ''
      - address: ''
  - type: ruby_proc
    columns:
      - name: location
        proc: |
          ->(_,record) do
            return record['Latitude'].to_s + "," + record['Longitude'].to_s
          end
        skip_nil: false
        type: string
      - name: address
        proc: |
          ->(_,record) do
            return record['Perfecture'].to_s + record['City'].to_s + record['AddressArea'].to_s
          end
        skip_nil: false
        type: string
  - type: column
    columns:
      - {name: LocationName, type: string}
      - {name: address, type: string}
      - {name: location, type: string}

#out: {type: stdout}
out:
  type: elasticsearch
  index: aedmap-tokyo-index3
  index_type: aed_schema
  nodes:
  - {host: 127.0.0.1}

出力結果のプレビューを確認しようとしたところ、以下のエラーが発生。

# embulk preview config.yml
～省略～
2019-10-04 07:15:40.495 +0200 [WARN] (0001:preview): Skipped invalid record (org.embulk.spi.DataException: com.jayway.jsonpath.InvalidJsonException: com.fasterxml.jackson.core.JsonParseException: Unexpected end-of-input: was expecting closing '"' for name
 at [Source: java.io.InputStreamReader@55adcf9e; line: 1, column: 34689])

データが悪いと言っているように見えたのですが、実際にembulk runコマンドで実行すると問題なく成功。理由が分からないのが気持ち悪いですが、そのまま作業を進めます。

[2019/10/4追記]
上のエラーが発生していたのはファイルサイズが大きいためでした。previewコマンドはデフォルトで32kbに設定されており、ファイルサイズが大きい場合はオプションで調整する必要がありました。
config.ymlの先頭に以下の2行を追加することで正常にpreviewを実行することができました。
（マニュアルをちゃんと確認しないと駄目ですね）

Configuration(Preview executor)

config.yml

exec:
  preview_sample_buffer_bytes: 4194304 # 512kb

実行結果は以下のようになります。

～省略～
+---------------------------+------------------------------+-------------------------+
|       LocationName:string |               address:string |         location:string |
+---------------------------+------------------------------+-------------------------+
|                 三鷹市東部市政窓口 |              東京都三鷹市中原1－29－35 |  35.6671705,139.5759333 |
|                     東台小学校 |              東京都三鷹市中原2－17－37 |  35.6673969,139.5720452 |
|                 東部水再生センター |                東京都三鷹市新川1－1－1 |  35.6679287,139.5784636 |
～省略～

previewコマンドで確認できたところ、Elasticsearchへのロードを実行してみます。

# embulk run config.yml

実行後、kibanaから「Management」-「Index Management」を選択すると、以下のようにインデックスが作成されており、「Docs count」でドキュメントが登録されていることが分かります。

Healthが"yellow"になっているのは、以下のコマンドを実行すると"green"に変わります。

# curl -H "Content-Type: application/json" -XPUT localhost:9200/*/_settings -d '{"number_of_replicas":0}'
{"acknowledged":true}

kibanaでの地図表示

データの投入も終わったので、いよいよkibanaで地図上にAEDの場所を表示させます。

最初に「Visualize」-「＋」を選択します。

次に「New Visualization」で「Coordinate Map」を選択します。

[Buckets]のAggregationに"GeoHash"、Fieldに"Location"を選択します。

「▶」を押すと、AEDの場所が地図にプロットされて表示されます。

説明は大分簡略化しているので、詳細は元記事を参照してください（素晴らしい元記事に「いいね」を！）

参考