REST API をPowershellで扱うための Elasticsearch のススメ (2)

前置き

前回の記事からの続きです。
以下のページにある内容をPowershellの Invoke-RestMethod コマンドレットで実施していきます。
https://qiita.com/math1101/items/311277789868ebd07835

なお、本記事はPowershell Ver.5.0向けです。3.0以降なら動きそうな気もしますが、2.0は存在しないコマンドレット多数、import-csvのencoding指定もNGのはず、ということで対象外としております。
Windows7のサポート期限も目前なので問題ないでしょう。

元データダウンロード

神戸市大気汚染常時監視結果 から灘大気測定局のデータを過去3年分（現在2018年なので2015年～2017年）ダウンロードします。
なお、観測局も年数も開始年も特に意味はありませんので、自由に変更いただいて構いません。

前回の記事と同様に、以下のコードをPowershellのプロンプトへ順次張り付けていきます。

# ■作業ディレクトリー作成
$WorkDir = "C:\Elastic\data"
if ( -not (Test-Path -Path $WorkDir) ) { # フォルダチェック
 New-Item -path $WorkDir -ItemType "directory"
}

# ■作業ディレクトリーへ移動
Set-Location $WorkDir

# ■ZIPファイルダウンロード
Function Get-TargetFile( $WorkDir, $URL ) {
    $FileName = Split-Path $URL -Leaf
    $FileFullPath = Join-Path $WorkDir $FileName
    Invoke-WebRequest -Uri $URL -Outfile $FileFullPath
}

Get-TargetFile -WorkDir $WorkDir -URL "http://kobe-taikikanshi.jp/kankyo/download/hour/kaku/2017_0103_hour_kakutei.zip"
Get-TargetFile -WorkDir $WorkDir -URL "http://kobe-taikikanshi.jp/kankyo/download/hour/kaku/2016_0103_hour_kakutei.zip"
Get-TargetFile -WorkDir $WorkDir -URL "http://kobe-taikikanshi.jp/kankyo/download/hour/kaku/2015_0103_hour_kakutei.zip"

# ZIPファイルの展開
$ZipList = Get-ChildItem $WorkDir | Where-Object {$_.Extension -eq ".zip" }
ForEach ($ZipFile in $ZipList) {
 Expand-Archive -Path $ZipFile.FullName -DestinationPath $WorkDir -Force
}

# ■展開後のファイル確認
Get-ChildItem $WorkDir

以上で該当ZIPファイルを展開して、以下のCSVファイルを取得するところまでできているはずです。

• 2015_0103_hour.csv
• 2016_0103_hour.csv
• 2017_0103_hour.csv

これで投入データの準備が整いました。

ちなみに、3ファイル以上取得する人のために即席で関数 Get-TargetFile を作ってます。
このように即席で関数が作れるのがPowershellのありがたいところです。

MAP作成

前回の記事でも軽く触れましたが、Elasticsearchでは、Json形式でデータを投入できます。
ただ、何も考えずにデータを投入すると、勝手にデータ型を文字列型にして登録します。
日付をDate型にしたいという場合には「Mapping」を先に行います。
データベースでいうところのテーブル定義（Create Table）です。

この作業は、一旦すべてを文字列として登録し、そのMAP情報を取得して、必要なところだけデータ型を変更して、それを正式なMAPファイルとするというのが一番手間が掛かりません。
MS SQL ServerでGUIでテーブルを生成してからSQL構文を生成するようなものですね。え？違う？？

ともかくやってみましょう。
以下、「■」のついている単位で順に実行してください。
（まとめて実行しても支障はありませんが、ステップ実行したほうが流れがくみ取れるかと思います）

# ■作業ディレクトリー再設定
$WorkDir = "C:\Elastic\data"
Set-Location $WorkDir

# ■Elasticsearch 登録先情報
$Hostname = "localhost"
$PortNumber = 9200
$IndexName = "air"
$TypeName = "ppm"
$BaseURL = "http://${Hostname}:${PortNumber}"
$APIURL = "/${IndexName}/${TypeName}/_bulk"
$RequestURL = $BaseURL + $APIURL
$ReqestType = "POST"
$BulkCommand = @{ "index" = @{ "_index" = $IndexName ; "_type" = $TypeName } } | ConvertTo-Json -Compress

# ■CSVファイル読み込み
$CsvList = Get-ChildItem $WorkDir | Where-Object {$_.Extension -eq ".csv" }
$CsvFile = Join-Path $WorkDir $CsvList[0]
$CsvData = Import-Csv -Path $CsvFile -Encoding Default

# ■CSV取り込み対象データ抽出
$FilterData = $CsvData | Where-Object { $_.'測定項目名称' -like "*${TypeName}*" } 
$FilterData.count

# ■Bulk実行用のJsonデータ生成
$BulkData = New-Object System.Collections.ArrayList # 新規配列
ForEach ($Counter in @(0..4) ) {
 $CsvRecord = $FilterData[$Counter]
 $TimeStamp = $CsvRecord."年" + $CsvRecord."月" + $CsvRecord."日" + "-" + $CsvRecord."時"
 $CsvRecord | Add-Member -NotePropertyName "TimeStamp" -NotePropertyValue $TimeStamp
 $JsonData = $CsvRecord | ConvertTo-Json -Compress
    $BulkData.Add($BulkCommand) > $Null
    $BulkData.Add($JsonData) > $Null
}

Write-Output $BulkData

# ■ Bulk実行
$JsonData = $( $BulkData -join "`n" ) + "`n"
$PostParam = [System.Text.Encoding]::UTF8.GetBytes($JsonData) # 日本語が文字化けするのでUTF8を強制
$Result = Invoke-RestMethod -Uri $RequestURL -Body $PostParam -ContentType 'application/json' -Method $ReqestType
$Result

# ■投入結果確認
$APIURL = "/${IndexName}/${TypeName}/_search"
$RequestURL = $BaseURL + $APIURL
$Result = Invoke-RestMethod -Uri $RequestURL
ConvertTo-JSON $Result -Depth 10

$APIURL = "/${IndexName}/${TypeName}/_count"
$RequestURL = $BaseURL + $APIURL
$( Invoke-RestMethod -Uri $RequestURL ).count

# ■現在のMAPを取得
$MapFile = Join-Path $WorkDir "map.json"
$APIURL = "/${IndexName}/${TypeName}/_mapping"
$RequestURL = $BaseURL + $APIURL
$Result = Invoke-RestMethod -Uri $RequestURL

ConvertTo-JSON $Result.air -Depth 10 | Out-File -Encoding Default -FilePath $MapFile

急に何を始めたのかというレベルになってきたので解説を挟みつつ。
まず、登録先URLの変数「$RequestURL」は「http://localhost:9200/air/ppm/_bulk」になります。
他のサイトなどでElasticsearchの解説を見ながら確認いただきたいですが、以下になります。

• index：データベースだとdatabaseに相当する概念
• type：データベースだとtableに相当する概念

そして「_bulk」がREST APIです。
index:air、type:ppmの中にまとめてデータを流し込む_bulk APIを使用するというのをURLで示している訳です。
以下のページのDocuments APIとか参考になるかもしれません。
https://medium.com/hello-elasticsearch/elasticsearch-api-83760ce1424b

そして、CSVファイルを読み込んで、登録対象「ppm」でフィルタリングします。
※ちなみに、ppmはperts per millionの略で体積濃度だそうです。門外漢のため細かいことは知りません。

続けて、MAP作成のひな型として、5件データを登録します。
フィルターしたCSVファイルから5件だけ情報を取得し、Timestanpというプロパティ（CSVの列に相当）を追加して、その1行をJson形式に変換します。

その後、_bulkのお作法に則ってJsonのリストを生成します。
index指定（$BulkCommand）で1行、投入データ内容（$JsonData）で1行の2行で1セットになっています。
_bulkのお作法は以下を参照してください。
https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html

そして、「■ Bulk実行」というところで5件のデータを登録しております。
Forで回しているときには配列に入れておき、Bulk実行直前でjoinで改行付きテキストに変換し、文字コードUTF8のバイトデータに置き換えてPOSTしております。
なお、_bulkに渡すJsonDataは、Forの中で直接テキストを生成しても問題ないです。配列に入れるのはForとかで回しやすいから癖で入れてるようなものです。

実行時にエラーが出なければ、投入結果確認のところで、投入されたデータと件数が表示されます。
問題がなければ、mapファイルを出力します。

このmapファイルをメモ帳等で開いてみましょう。
ちなみに、おすすめはVisual Studio Codeです。「ドキュメントのフォーマット」機能でJsonファイルを見やすく整えてくれます。エンコード指定しないとすぐに化けるのが難点ですが。

このファイルのTimeStampを編集します。
Before:

                "TimeStamp": {
                    "type": "text",
                    "fields": {
                        "keyword": {
                            "type": "keyword",
                            "ignore_above": 256
                        }
                    }
                },

After:

                "TimeStamp": {
                    "type":"date",
                    "format":"YYYYMMdd-HH"
                },

長くなったので続きは次の記事とします。
ただ、この記事の中で、本来の目的である Invoke-WebRequest によるREST APIの操作方法はなんとなく理解できたと思いますので、それが目的の人はここで離脱すればいいと思います。
せっかくなんでCSV全データ投入まで見てやるぜという人はもう少しお付き合いください。