# Elasticsearch内のデータの結合について

**URL:** <https://discuss.elastic.co/t/elasticsearch/175429>\
**Category:** 日本語による質問・議論はこちら\
**Created:** [April 4, 2019, 1:46pm UTC](https://discuss.elastic.co/t/elasticsearch/175429 "2019-04-04T13:46:56Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![iasenust](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/iasenust/32/43502_2.png) [@iasenust](https://discuss.elastic.co/u/iasenust)\
**Post date:** [April 4, 2019, 1:46pm UTC](https://discuss.elastic.co/t/elasticsearch/175429/1 "2019-04-04T13:46:56Z")

</div>

初心者の者です。  
Aというログには、「コマンドID、コマンド名」Bというログには、「コマンドID、コマンド応答」が多数記載されています。  
それぞれのログをlogstashから取り込んで、「コマンドID、コマンド名、コマンド応答」  
というドキュメントにしたい場合、どのようにすればよいのでしょうか？  
ログを取り込む前に、kibanaのdevtoolsに何か設定しておいてうまくできるのでしょうか？  
基本的なことで申し訳ありませんがご教授ください。

---

<div class="post-metadata">

**Author:** ![tsgkdt](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tsgkdt/32/39151_2.png) [@tsgkdt](https://discuss.elastic.co/u/tsgkdt)\
**Post date:** [April 4, 2019, 5:05pm UTC](https://discuss.elastic.co/t/elasticsearch/175429/2 "2019-04-04T17:05:13Z")

</div>

こんにちわ。

コマンドIDをキーとして、ログAとログBが1対1で紐づいている、ということでしょうか。  
ということであれば、Logstashの中でOutputのフィルタリングにelasticsearchを指定しているかと思いますが、  
ここで、 doc\_as\_upsertという設定ができるので、これをtrueにしてみてください。  
[https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-doc\_as\_upsert](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-doc_as_upsert)

こうすることで、IDをキーに１つのドキュメントに対して、それぞれAとBのログをいれると、「コマンドID、コマンド名、コマンド応答」のようなドキュメントとなります。  
AのあとにBが入っても、先にBがあって後からAでも、IDさえあればマージされたように見えるはずです。

以下こちらで試したところを貼っておきます。

## テストデータ

２つのCSVを用意する。それぞれ、a.csv, b.csvとする。

- a.csv

```auto
ID,NAME
1,コマンド名Ａ
2,コマンド名Ｂ
3,コマンドー

```

- b.csv

```auto
ID,VALUE
1,成功
2,失敗
3,成功

```

## Logstashの設定

ポイントはOutputのところでUpsertを指定してところです。  
document\_idがなければinsert、あればupdateとなります。  
既存の項目値はそのまま上書されてしまいますが、今回のようにNAMEとVALUEとかぶらないフィールドであればマージされたような状態となり期待した結果となるように思います。

```auto
input {
  file {
      path => "/tmp/a.csv"
      start_position => "beginning"
      tags => ["type_a"]
  }

  file {
      path => "/tmp/b.csv"
      start_position => "beginning"
      tags => ["type_b"]
  }
}

filter {

   if "type_a" in [tags] {

       csv {
            columns => ["id","name"]
            separator => ","
            skip_header => true
       }
   } else {

       csv {
            columns => ["id","value"]
            separator => ","
            skip_header => true
       }
   }

   
}

output {
    elasticsearch {
        hosts => ["elasticsearch:9200"]
        index => "forum0404"
        document_id => "%{id}"
        doc_as_upsert => true
        action => "update"
    }
}

```

## 結果

```json
{
        "_index" : "forum0404",
        "_type" : "_doc",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "@version" : "1",
          "tags" : [
            "type_b"
          ],
          "@timestamp" : "2019-04-04T16:48:41.302Z",
          "host" : "bd72215c114b",
          "id" : "3",
          "name" : "コマンドー",
          "message" : "3,成功\r",
          "path" : "/tmp/b.csv",
          "value" : "成功"
        }
      }

```

## 確認した環境

- Elasticserch: 7.0.0-rc2 (on Docker)
- Logstash: 7.0.0-rc2 (on Docker)
- Kibana: 7.0.0-rc2 (on Docker)

## 参考情報

- [Upsert with logstash](https://discuss.elastic.co/t/upsert-with-logstash/59116)

---

<div class="post-metadata">

**Author:** ![iasenust](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/iasenust/32/43502_2.png) [@iasenust](https://discuss.elastic.co/u/iasenust)\
**Post date:** [April 5, 2019, 1:23am UTC](https://discuss.elastic.co/t/elasticsearch/175429/3 "2019-04-05T01:23:07Z")

</div>

早急なご回答、ありがとうございました。  
ご教授いただいた通りにやったらうまくいきました。大変助かりました。

一つ興味でお聞きしたいのですが、Referenceマニュアルのdoc\_as\_upsertの記載を見ると  
updateモードを有効にする、みたいなことが書いてあるのでdoc\_as\_upsertだけ追記すればいい  
のかな、と思い、action=\>の行をはじめ省いてやってみたのですが、うまくできませんでした。  
このあたりのマニュアルからは読めないところをどのように習得していっていますか？

よろしくお願いいたします。

---

<div class="post-metadata">

**Author:** ![tsgkdt](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tsgkdt/32/39151_2.png) [@tsgkdt](https://discuss.elastic.co/u/tsgkdt)\
**Post date:** [April 5, 2019, 4:38am UTC](https://discuss.elastic.co/t/elasticsearch/175429/4 "2019-04-05T04:38:15Z")

</div>

すみません、action =\> "update" のところについて言及しなかったのは、言葉足らずでした。

さて、公式マニュアルだけじゃよく分からん！というときどうするか、です。  
私なら、という私見になりますがご容赦ください。

確実なのは・・・

1. Elastic社のトレーニングに行き質問をたくさんしてくる。（ちょうど5、6月にトレーニングあります）
2. サブスクリプションをつけて、サポートの人に聞く

直近のトレーニング予定はこちら（日本語で研修受けられます）  
[https://training.elastic.co/instructor-led-training/ElasticsearchEngineerI](https://training.elastic.co/instructor-led-training/ElasticsearchEngineerI)  
なお、回し者ではありません。

他には・・・私が思いつく習得方法です。

1. Elastic社のイベントにいき Ask me Anythingブースで質問しまくる（本当に何でも聞いて良いらしい）
2. Elasticsearchの勉強会にいき、懇親会で人を見つけて聞いてみる（どうやって情報収集していますか？ こういうことで困ってるんですよ、とか）
3. DiscussやGitHubのissueで類似の事例がないかどうかを”検索”する（なければ、スレッドをたてて実際に聞いてみる）
4. 自分で仮説を立てて試行錯誤してみる
5. OSSの強みを活かして、ソースを読む

コミュニティに参加したり頼ったりすることで、習得が早まると思いますです。

---

<div class="post-metadata">

**Author:** ![iasenust](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/iasenust/32/43502_2.png) [@iasenust](https://discuss.elastic.co/u/iasenust)\
**Post date:** [April 6, 2019, 2:42am UTC](https://discuss.elastic.co/t/elasticsearch/175429/5 "2019-04-06T02:42:10Z")

</div>

どうもありがとうございます。  
やはり公式マニュアルを読むだけでなく、有識者からノウハウを伝授頂かないとですね。  
参考になります。

---

<div class="post-metadata">

**Author:** ![iasenust](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/iasenust/32/43502_2.png) [@iasenust](https://discuss.elastic.co/u/iasenust)\
**Post date:** [April 16, 2019, 8:26am UTC](https://discuss.elastic.co/t/elasticsearch/175429/6 "2019-04-16T08:26:16Z")

</div>

お世話になっております。  
ご教授いただいた内容から、以下を新たに実現させたいと思っています。

・コマンドIDが同じでかつ、ログ内のタイムスタンプの「時」が1だけずれている際に、何か別の同一のIDにして紐づけたい

上記を要望する背景ですが tsgkdt様の例でいうと、a.csvとb.csv内のIDはログの中で使いまわされており、データを一意には紐づけられないことがわかりました。また、二つのログは要求ログ、応答ログの関係にあり、ログ内のタイムスタンプが最大数十秒ずれており、タイムスタンプのみをIDにして括り付けることもできないことがわかっています。  
そこで、  
①IDが同じである  
②タイムスタンプがずれていてもせいぜい「時」が1ずれているだけ  
というのを条件にして括り付ければよい、という条件を設定しました。  
が、「すでにDBに登録されている値とこれから入れようとしている値を比較して差が1以下であれば・・・」のようなif文をlogstash? のconfigに入れ込むことは可能なのでしょうか？  
※logstashのconfigはあくまでElasticsearchに入れる際の条件を書いているので、そのようなことは  
できなそうだな・・・・と思っているのですが。

よろしくお願いいたします。

---

<div class="post-metadata">

**Author:** ![tsgkdt](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tsgkdt/32/39151_2.png) [@tsgkdt](https://discuss.elastic.co/u/tsgkdt)\
**Post date:** [April 19, 2019, 8:21am UTC](https://discuss.elastic.co/t/elasticsearch/175429/7 "2019-04-19T08:21:15Z")

</div>

予めどちらかのデータがインデックス済みで、検索可能状態になっている、という制限はありますが、  
こんなアプローチも考えられそうです。

## 前提

1. ２つのインデックスは、共通のIDというフィールドを持っており、値は共通である
2. IDとタイムスタンプが±１秒の範囲内という条件で、一意に対となるデータがみつかる

filterの中でElasticsearchにクエリを投げ、その結果を取得してデータを加工することができます。

[https://www.elastic.co/guide/en/logstash/current/plugins-filters-elasticsearch.html](https://www.elastic.co/guide/en/logstash/current/plugins-filters-elasticsearch.html)

## 考えられるアプローチ

後からデータを入れる方が、先に入っているデータを”検索して”、その結果をもとにインデックスを登録するようにする。

IDが同じで、timestampで前後1秒の時間のズレ、というものがありますが、これをクエリで表現して結合先の対象データを取得できれば良いと考えます。

Logstashのfilter部分をこのように書いて、template.jsonを使ってクエリを渡すようにします。

```ruby

filter {
  elasticsearch {
  
    hosts => ["xxx.xxx.xxx.xxx:9200"]
    index => "testtesttest"
    query_template => "template.json"
    # メタ情報
    docinfo_fields => {
      "_id" => "document_id"
      "_index" => "document_index"
    }
    #取得した検索結果から、値を取り出してマッピングする
    fields => {
      "code" => "code"
    }
  }
}

```

template.jsonでは、IDが同じ、前後1秒のズレを考慮したクエリを書きます。

```json

{
    "size": 1,
    "sort": [
        {
            "@timestamp": "desc"
        }
    ],
    "query": {
        "bool": {
            "must": [
                {
                    "range": {
                        "@timestamp": {
                            "gte": "%{[target_date]}||-1s",
                            "lt": "%{[target_date]}||+1s",
                            "time_zone": "+00:00"
                        }
                    }
                },
                {
                    "term": {
                        "_id": {
                            "value": "%{[target_id]}"
                        }
                    }
                }
            ]
        }
    },
    "_source": [
        "code"
    ]
}

```

rangeクエリのところを見ていただくと、%{[フィールド名]} があり、ここにLogstashで置換される部分です。  
後ろに`||+1s` や `||-1s`と書くことで、前後1秒を求めています。

もし置換文字列でなければ、このようなイメージです。

```json
 "range": {
     "@timestamp": {
         "gte": "2019-04-16 21:00:10||-1s",
          "lt": "2019-04-16 21:00:10||+1s",
          "time_zone": "+00:00"
     }
}

```

このアプローチですと、Elasticsearchで検索できることが前提なので、既にデータが入っていて検索可能となっていなければなりません。  
また、場合によっては多量のクエリが発行されることもあるため、必ずしも良いやり方とは言えないでしょう。

時間をずれないようにするとか、もう少し手前の処理で２つのファイルを統合するような処理を別に検討するとか考えても良いように思います。

---

<div class="post-metadata">

**Author:** ![iasenust](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/iasenust/32/43502_2.png) [@iasenust](https://discuss.elastic.co/u/iasenust)\
**Post date:** [April 23, 2019, 12:34am UTC](https://discuss.elastic.co/t/elasticsearch/175429/8 "2019-04-23T00:34:06Z")

</div>

詳細にご回答頂いて、大変助かります。  
レベルが高い内容なので、時間をかけて試していこうと思います。  
ありがとうございます。

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [May 21, 2019, 12:34am UTC](https://discuss.elastic.co/t/elasticsearch/175429/9 "2019-05-21T00:34:07Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.