netatomのstationデータをAWS Elasticsearch Serverへ流し、そのデータ描画をする #Python

この記事は何？

前の記事の続きとなります。
netatmoのデータ取得ができたので、このデータを使ってAWS Elasticsearch Serverでデータ取得から描画まで行います。

AWS Elasticsearch Serviceとは？

概要だけつまむと、データレイクのElasticsearchと可視化BIツールのkibanaが合体したサービス。各サービスの結合と必要な設定を担当してくれます。

AWS Elasticsearch Serviceの環境構築

AWS Elasticsearch Serviceの新規環境構築していきます。

AWSコンソールを開きElasticsearch Serviceから新しいドメイン作成から環境構築します。

途中の設定のうち今回は動くことを目標とするため、デフォルトから２つ設定を変更します。
１つがネットワーク構築です。これをパブリックアクセスとすることでどこからでもアクセスが可能と変更します。
このままだとどこからでもアクセスし放題のため、次でアクセス制限をかけます。

２つ目のアクセス制限です。アクセスポリシーからCIDR表記でアクセス制限をかけます。

これから最後まで設定して、最終的に新しくドメインを取得することができ、elasticsearchとkibanaの環境が得られました。

Elasticsearchにデータを挿入する

前章で取得したnetatmoのアクセストークンを使ってデータをkibanaに流し込みます。
この時ポイントは２つあり、netatmoのデータを加工しElasticsearchが時系列として扱えるよう日付データを追加し、kibanaのエンドポイントを都度変更しデータの上書きをしないようにしています。

時系列データを扱うため、このプログラムをlambdaで定期実行にしたり、cronで定期実行にしたりして時系列として扱えるようにします。

import datetime
import json
import requests
from requests.auth import HTTPBasicAuth


def get_date_suffix():
    now = datetime.datetime.now()
    return now.strftime('%Y%m%d%H%M%S')


def get_station_data():
    netatmo_headers = {'Content-Type': 'application/x-www-form-urlencoded', 'charset': 'UTF-8'}
    netatmo_url = 'https://api.netatmo.com/api/getstationsdata'
    netatmo_data = {
        'access_token': 'アクセストークン'
    }

    resp = requests.post(url=netatmo_url, headers=netatmo_headers, data=netatmo_data)
    json_data = json.loads(resp.text)

    # netatmoのデータがタイムスタンプ形式から、Elasticsearchで扱える形である「yyyy/mm/dd hh:mm:ss」へと変換し、カラムを追加する。
    unix_time = datetime.datetime.fromtimestamp(json_data['body']['devices'][0]['dashboard_data']['time_utc'])
    json_data['body']['devices'][0]['dashboard_data']['@timestamp'] = unix_time.strftime("%Y/%m/%d %H:%M:%S")

    resp_data = json.dumps(json_data)
    return resp_data

# エンドポイントを上書きしないように、リクエストのたびにURIの末尾を変更している
url = 'AWS Elasticsearch Service記載のエンドポイント' + '/trg/_doc/trg-kawahara' + get_date_suffix()
headers = {
    'Content-Type': 'application/json'
}
data = get_station_data()
auth = HTTPBasicAuth('kibanaのユーザーID', 'kibanaのパスワード')

basic_auth_req = requests.post(url=url, headers=headers, data=data, auth=auth)
print(basic_auth_req.text)

レスポンスは下のようにsuccessfulが帰ってきます。

{
  "_index": "trg",
  "_type": "_doc",
  "_id": "trg-kawahara20210816192617",
  "_version": 1,
  "result": "created",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "_seq_no": 0,
  "_primary_term": 1
}