ELK Stackの利用6：半年間分データ投入

前回までの記事：

ELK Stackの利用1：はじめに

ELK Stackの利用2：ログ自動取得

ELK Stackの利用3：ログ自動取得その2

ELK Stackの利用4：ESデータ投入（Logstash編）

ELK Stackの利用5：ESデータ投入（Bulk API編）

 

まず最初にこの話題から。ESクラスタのノードはVMware上のVMである。VMwareはデフォルトで”ハイパースレッド機能”が「有効化」になっている。前回の一日分のログ投入時間が30分というのは、この状態で実施したものだった。このハイパースレッドを「無効化」にしたところ、なんと、10分程度で投入が終了した。どうもこの機能を有効にすると、CPUを目一杯使う場合、たとえばBulk APIによるインデクシングのような場合は悪い方へ働くようだ

半年分データの投入の前に試験環境の構成が変わった。VMware上の3ノード構成はそのままだが、Dockerコンテナに各ノードを置き、全体をKubernetesで管理する形になった。各ノードは4CPU、8GBメモリ、140GB SSD、16TB HDDで構成。OSはCentOS7、そして、ESのバージョンは7に上げた。もちろんハイパースレッド機能はオフである。

そしてもう一つ。今までインデクシングの前作業、インデックス作成やデータの加工等のために、いくつかスクリプトを作ってきた。8日間程度のデータ量ではその実行時間は気にならなかったが、半年間分データでは状況が異なるため、スクリプトの見直しを行った。まず、シェルスクリプト内の配列の利用をやめたところ、2分かかっていたものが1分30秒に縮まった。続いて別個に実行していたスクリプトを一つにまとめ、一回のループで処理するようにした。さらに、pythonスクリプトでcsvからjsonへ変換していたが、ここがかなり時間を食っていたため、下記のようにC++に書き変えた。これにより、pythonでかかっていた時間が1/3に削減できた。

/*-----------------------------------------------
# chg_csv2json.cpp
#    Convert csv format files to json.
-----------------------------------------------*/

#include <iostream>
#include <string>
#include <fstream>
#include <sstream>
#include <vector>

using namespace std;

/*-------- split --------*/
vector<string> split(string& input, char delimiter)
{
    istringstream stream(input);
    string field;
    vector<string> result;
    while (getline(stream, field, delimiter)) {
        result.push_back(field);
    }
    return result;
}

/*-------- Main Routine --------*/
int main(int argc, char** argv)
{
  ifstream ifs;
  ofstream ofs;
  string csvfile,jsonfile,indexname;
  string indexstr,line,buf;
  int pos;

  // Get csv filename and indexname
  if(argc != 3){
    cout << "Usage: " << argv[0] << " <filename> <indexname>" << "\n";
    return 0;
  }

  csvfile = jsonfile = argv[1];
  indexname = argv[2];

  // Write json filename by replacement a suffix: ".csv"
  pos = (int)jsonfile.rfind(".");
  jsonfile.replace(pos,4,".json");

  // Create a text including indexname
  indexstr = "{\"index\":{\"_index\":\""+indexname+"\"}}";

  // Open csvfile for input
  ifs.open(csvfile.c_str());
  if(!ifs){
    cout << "Error: Cannot open file: [" << csvfile << "]\n";
    return 0;
  }
 
  // Open jsonfile for output
  ofs.open(jsonfile.c_str());

  // Read file
  // Get the first line (header) and split them to refer by an array
  getline(ifs,line);
  vector<string> f_array = split(line, ',');    // Save fields 

  // Get from the second line to the end
  // Create a string like "field1":"data1","field2":"data2",...  
  while(getline(ifs,line)){
    vector<string> d_array(f_array.size(),"");  // Get area d_array is expected
    d_array = split(line, ',');                 // Save data
    buf = indexstr + "{" + "\"" + f_array[0] + "\": \"" + d_array[0] + "\"";
    for(int i=1; i<f_array.size(); i++){
      buf += ", \"" + f_array[i] + "\": \"" + d_array[i] + "\""; 
    }
    // Write to output
    buf += "}\n";
    ofs<<buf;
  }

  // Close files
  ifs.close();
  ofs.close();
}

ES投入前段階ではこういう作業をする。かっこ内は、一日分データを処理した時のおおよその所要時間。全部まとめて10分以下になった。

1. もとデータファイルのcsv化と、フィールド名とその型の抽出（4分10秒）
2. インデックス用スクリプト作成（45秒）
3. csvをjsonフォーマットに変換（2分30秒）
4. jsonファイルをES投入に向け加工（2分10秒）

一通り準備ができたので、半年分データ（4/1〜9/30）のインデクシングに進んだ。ES投入もスクリプトで実行したが、本体は while read file; do echo $file; curl -XPOST 'http://log.xxx.yyy.jp:9200/_bulk' -s -H 'Content-Type: application/x-ndjson' --data-binary "@$file" > $file.log; done < all.list で、all.listに投入する全jsonファイル名を記述している

実際の投入は様子を見ながら少しづつ進めた。エラーもそれなりに出たため、夜に投入して翌朝投入結果ログを確認、エラーに対応、という日課になっていた。4/1〜4/30までのES投入は、一日分ログのインデクシング時間が平均19分だった。そのうちにエラーもだいぶ潰されたので投入もスムーズになっていき、二週間程で4/1〜8/27のインデクシング終了。8/28データの投入で、このエラーが出た。

root_cause":[{"type":"illegal_argument_exception","reason":"Validation Failed: 1: this action would add [2] total shards, but this cluster currently has [15300]/[15300] maximum shards open;""

シャード数越えエラーだった。シャードとはクラスタにデータを分散するときの単位。ここでは1シャード1インデックス（＝日別のファイル種別毎）で、ファイル数が増えれば、シャード数も大きくなる。シャード数の変更もcurlを使う。半年分データでは、最終的に6900に設定してうまくいった。

$ curl -XPUT 'log.xxx.yyy.jp:9200/_cluster/settings' -H 'Content-type: application/json' --data-binary $'{"transient":{"cluster.max_shards_per_node":6900}}'
{"acknowledged":true,"persistent":{},"transient":{"cluster":{"max_shards_per_node":"5400"}}}

これで8/28〜9/30のログデータ投入も済み、半年間のデータ投入完了。スケールが大きくなるのでもっといろいろあるかと思ったが、意外にすんなりここまで来られた。なお現在のシャード数は以下で確認できる。これはその後さらにデータを投入した後で調べたときのもの。

$ curl -XGET 'http://log.xxx.yyy.jp:9200/_cat/health?v'
epoch      timestamp cluster status node.total node.data shards   pri relo init unassign pending_tasks max_task_wait_time active_shards_percent
1592965086 02:18:06  datalog green           3         3  25618 12809    0    0        0             0                  -                100.0%

shardsにはクラスタ全体のシャード数が表示され、今は3ノード使っているので、1ノードあたりのシャード数は8539になる。このコマンドは実はESのヘルスチェックをするもので、status=green、active_shards_percent=100％であれば、ESとして問題なく機能していることを示す。

ところで、半年分のデータはSSDに投入していた。そして、この半年分のデータでSSDを9割方使ってしまったため、SSDに入れたデータは大容量のHDDへ移され、以後はHDDへ直接投入することになった。さらに三ヶ月分追加投入。javaヒープサイズやシャード数を多くしたりなどして成功。今までインデクシング時間は20分弱だったのだが、HDDはやはり遅く、投入時間は1.5倍に増えた。

さらにもう一ヶ月分、というところで問題発生。ヘルスチェックで"yellow"が出た。

$ curl -XGET 'http://log.xxx.yyy.jp:9200/_cat/health?v'
epoch      timestamp cluster status node.total node.data shards   pri relo init unassign pending_tasks max_task_wait_time active_shards_percent
1594951163 01:59:23  datalog yellow          3         3  26311 14989    0    2     3665             0                  -                 87.8%

ESの各ノードを調べたところ、いずれもメモリを9割程使用。さらに、シャード数が多すぎ、かつ各シャードサイズが小さすぎる、ということがわかってきた。日別ファイル種別ごとに1インデックス、としていたことが裏目に出た。まず多すぎるインデックスがメモリを食っていた。そしてシャードサイズはElasticsearchの推奨が20〜40GB、けれど我々の環境では最大で600MB程度で、これはオーバーヘッドを高くし、システムパフォーマンスを悪くするそうだ。

というわけで、設計から見直さなければならなくなった。1インデックスを日別でなく月別にする方針に変更。スクリプトの変更は、インデックスの作成とES投入時のインデックス名を変更するだけで大きなものではなかったのだが、この試験についてはここで終了になった。すでにESに投入されたデータの扱いに話がシフトしていったことと、本務の仕事が忙しくなりこの作業を続けられなくなったことのためである。中途半端で残念だったが、本務ではないし致し方ない。ESについていろいろわかったことで良しとした。

 

2024/12/18（水）Blog 21