.jsonファイルの書式ごとに、Logstash経由でのElasticsearchへの取り込み結果をまとめてみた。

動作環境

OS：Ubuntu Server 20.04.2 LTS
Elasticseach：7.13
Logstash：7.13

logstashの起動

.jsonファイルの取り込みを試すときは、
systemctl でやるよりも、binを直接叩いて実行した方がやりやすい。
systemctlだと、ログを都度journalctlで見る必要があり、面倒。
.binだと、叩いた直後からログがコンソールに出てくれる。

ログを見ないと、
「設定ファイルをちゃんと読み込んでない」のか、
「設定ファイルの内容が不適切で、取り込み対象のファイルを無視している」のかわかりにくい。
感覚的に詰まったときは大体後者。

binで実行する場合

sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/hoge.conf

systemctlで実行する場合

systemctl start logstash.service

# logの確認
journalctl -u logstash.service -xe -n 100

.jsonの書式別　取り込み結果

■ .json(配列 + 改行 \n なし)　... Elasticseacrchへの取り込み〇

hoge.json

[{"name":"toda","age":24},{"name":"shimoda","age":28},{"name":"kasuya","age":34}]

取り込みやすい形。

hoge.conf

input {
    file {
        path =>[/data/hoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
        codec => "json"
    }
}

#filter {
#
#}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

出力(Elasticsearchでの見え方)

～ ドキュメント1 (DBでいうレコード） ～
・・・
"name":"toda",
"age":24,
・・・
～ ドキュメント2 (DBでいうレコード） ～
・・・
"name":"shimoda",
"age":28,
・・・
～ ドキュメント3 (DBでいうレコード） ～
・・・
"name":"kasuya",
"age":34,
・・・

期待値

####■ .json(配列 + 改行 \n あり)　... Elasticseacrchへの取り込み×

hoge.json

[
    {
        "name":"toda",
        "age":24
    },
    {
        "name":"shimoda",
        "age":28
    },
    {
        "name":"kasuya",
        "age":34
    }
]

hoge.conf

input {
    file {
        path =>[/data/hoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
        codec => "json"
    }
}

#filter {
#
#}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

.json(改行 \n なし)の場合と同じ

出力

例）"message" : "}",
　　"message" : """ "name":"toda",""",

\nがあるため、取り込めない。
一行ずつのレコードができる。
それぞれのレコードのmessageに、一行ごとのこまごまになった値が入る。

取り込もうと思えばできるけど、単純なオプション設定ではできないため、
細かくパースする設定が必要そう。
.jsonファイルを取り込む前にjqコマンドなどで一行の.jsonにした方が楽そう。

jq --compact-output

■ .json(配列ではない + 改行 \n なし)　... Elasticseacrchへの取り込み〇

hoge.json

{"name":"toda","age":"24"}

hoge.conf

input {
    file {
        path =>[/data/hoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
        codec => "json"
    }
}

filter {

}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

出力

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "hogeIndex",
        "_type" : "_doc",
        "_id" : "Xa9xP3sBW11KQnuE-Tml",
        "_score" : 1.0,
        "_source" : {
          "host" : "HOGEHOGEHOGEHOGE",
          "age" : "24",
          "@timestamp" : "2021-08-17T12:17:19.067Z",
          "@version" : "1",
          "path" : "/data/hoge.json",
          "name" : "toda"
        }
      }
    ]
  }
}

期待値通りに取り込んでいる。

別の.confだと、

hoge.conf

input {
    file {
        path =>[/data/hoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
    }
}

filter {
    json {
        source => "message"
    }
}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

出力

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "hogeIndex",
        "_type" : "_doc",
        "_id" : "XK9uP3sBW11KQnuEGznB",
        "_score" : 1.0,
        "_source" : {
          "host" : "HOGEHOGEHOGEHOGE",
          "@version" : "1",
          "path" : "/data/hoge.json",
          "name" : "tanaka",
          "message" : """{"name":"tanaka","age":"23"}""",
          "age" : "23",
          "@timestamp" : "2021-08-17T12:13:05.646Z"
        }
      }
      ...
    ]
  }
}

この方法の場合は、
nameとageがそれぞれ別のフィールドとして取り込めているが、
messageフィールドに取り込みデータがそのまま入り、データが重複している。
.confのfilterでmessageフィールドをremoveする必要がある。

ただし、以下の形式はいまいち

{"name":"toda","age":24},{"name":"shimoda","age":28},{"name":"kasuya","age":34}

hoge.conf

input {
    file {
        path =>[/data/hoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
        codec => "json"
    }
}

filter {

}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

出力

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "hogeIndex",
        "_type" : "_doc",
        "_id" : "Xa9xP3sBW11KQnuE-Tml",
        "_score" : 1.0,
        "_source" : {
          "host" : "HOGEHOGEHOGEHOGE",
          "age" : "24",
          "@timestamp" : "2021-08-17T12:17:19.067Z",
          "@version" : "1",
          "path" : "/data/hoge.json",
          "name" : "toda"
        }
      }
    ]
  }
}

{"name":"toda","age":24},{"name":"shimoda","age":28},{"name":"kasuya","age":34}　のうち、
最初の{"name":"toda","age":24}しか取り込まれなかった。

別の.confの場合

hoge.conf

input {
    file {
        path =>[/data/hoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
    }
}

filter {
    json {
        source => "message"
    }
}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

出力

...
"message" : """{"name":"toda","age":24},{"name":"shimoda","age":28},{"name":"kasuya","age":34}""",
...

一つのmessageフィールドにすべて含まれていた。
ドキュメント（DBでいうレコード）は一つのみ。
nameフィールドも、ageフィールドも作成されなかった。
つまりダメ。

####■ .json(配列ではない + 改行 \n あり)　... Elasticseacrchへの取り込み×

hoge.json

{
    "name":"toda",
    "age":24
},
{
    "name":"shimoda",
    "age":28
},
{
    "name":"kasuya",
    "age":34
}

や

hoge.json

{
    "name":"toda",
    "age":24
}

hoge.conf

input {
    file {
        path =>[/data/hoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
        codec => "json"
    }
}

#filter {
#
#}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

や

hoge.conf

input {
    file {
        path =>[/data/hogehoge.json],
        sincedb_path => "/dev/null"
        start_position => "beginning"
        codec => "json"
    }
}

filter {

}

output {
    elasticsearch {
        hosts => ["http://X.X.X.X:9200"]
        index => "hogeIndex"
    }
}

出力

...
"message" : """ "name":"toda",""",
...

messageフィールドのみ。ダメ。

##　結論
とりあえず取り込む.jsonファイルは改行\nなしのものにする。
一つのファイルに配列として足す形式か、

hoge.json

[{"name":"toda","age":24},{"name":"shimoda","age":28},{"name":"kasuya","age":34}]

データごとにファイルを分ける以下の形式にするか、にしようと思う。

hoge1.json

{"name":"toda","age":"24"}

hoge2.json

{"name":"shimoda","age":"28"}

hoge3.json

{"name":"kasuya","age":"34"}

##　メモ
.confでcodec=>"json_lines"で改行ありも取り込めるっぽいけど、
何回か試して出力が不安定（どういう形式の.jsonが正なのかよくわからなかった）ので、
json_linesの使用は諦めた。