【Logstash】不要な項目（フィールド）を削除してElasticsearchのデータ容量を節約する

2021年12月21日
2023年3月6日
Elastic
elasticsearch, logstash

Elastic

LogstashはElasticsearchにデータを投入するためのツールで、リアルタイムに生成されるデータを随時加工し、Elasticsearchに送出することができます。

Logstashを使ってデータを入れる際、全ての項目を入れなくてよいケースもあると思います。不要なデータを削ることは、容量を節約するうえでも重要です。

今回は、Logstashからデータを投入する際に不要な項目（フィールド）を削除する方法を紹介します。

1 使用するデータ
2 そのままデータを投入する
3 フィールドの削除
4 まとめ

使用するデータ

今回はGoogleの株価データを使用します。

Date	High	Low	Open	Close	Volume	Adj Close
2015/11/23	2095.61	2081.39	2089.41	2086.59	3587980000	2086.59
2015/11/24	2094.12	2070.29	2084.42	2089.14	3884930000	2089.14
2015/11/25	2093	2086.3	2089.3	2088.87	2852940000	2088.87
2015/11/26	2093	2086.3	2089.3	2088.87	2852940000	2088.87
2015/11/27	2093.29	2084.13	2088.82	2090.11	1466840000	2090.11
2015/11/28	2093.29	2084.13	2088.82	2090.11	1466840000	2090.11
2015/11/29	2093.29	2084.13	2088.82	2090.11	1466840000	2090.11
2015/11/30	2093.81	2080.41	2090.95	2080.41	4275030000	2080.41

google_stock.csv

そのままデータを投入する

まずは項目削除を一切行わずにElasticsearchにデータを入れてみます。使いやすいようにデータの加工はしています。これらの処理は今回の内容とは関係ありません。

input {
  file{
	path => "C:/Project/*_stock.csv"
	sincedb_path => "C:/Project/stock_sincedb"
	start_position => "beginning"
	}
}


filter{

	csv{
		skip_header => "true"
		columns => ["Date", "High", "Low", "Open", "Close", "Volume", "Adj", "close"]
		separator => ","
	}
	
	mutate{
		convert => {
			"High" => "float"
			"Low" => "float"
			"Open" => "float"
			"Close" => "float"
			"Volume" => "float"
			"Adj" => "float"
			"close" => "float"
		}
	}

	grok{
		match => { "path" => "%{GREEDYDATA:path}/%{GREEDYDATA:company}_stock.csv"}
	}

	date{
		match => ["Date", "yyyy-MM-dd" ]
	}
}

output {
  elasticsearch {
    hosts => "http://localhost:9200"
    index => "%{company}_stock-%{+YYYY}"
  }
}

データを確認してみると、

csvファイルにあった項目はもちろん、hostやmessageなど心当たりのないフィールドも勝手に追加されてしまっています。

今回はhostなど勝手に追加されたフィールドのほか、もともとcsvファイルにあったAdjを削除してみます。

フィールドの削除

フィールドの削除はremove_fieldで行い、filter領域のmutateの内部に記載します。具体的には以下のように書きます。

filter{
　mutate{
　　remove_field => ["message", "host", "Adj"]
　}
}

上記では、”message”、”host”、”Adj”を削除するように設定しています。

これを元々の設定ファイルに組み込むと以下のようになります。

input {
  file{
	path => "C:/Project/*_stock.csv"
	sincedb_path => "C:/Project/stock_sincedb"
	start_position => "beginning"
	}
}


filter{

	csv{
		skip_header => "true"
		columns => ["Date", "High", "Low", "Open", "Close", "Volume", "Adj", "close"]
		separator => ","
	}
	
	mutate{
		convert => {
			"High" => "float"
			"Low" => "float"
			"Open" => "float"
			"Close" => "float"
			"Volume" => "float"
			"Adj" => "float"
			"close" => "float"
		}
		
		remove_field => ["message", "host", "Adj"]
	}


	grok{
		match => { "path" => "%{GREEDYDATA:path}/%{GREEDYDATA:company}_stock.csv"}
	}

	date{
		match => ["Date", "yyyy-MM-dd" ]
	}
}

output {
  elasticsearch {
    hosts => "http://localhost:9200"
    index => "%{company}_stock-%{+YYYY}"
  }
}

この設定で投入されたデータを見てみます。