# Aggregateフィルタについて。重複してしまいます。

**URL:** <https://discuss.elastic.co/t/aggregate/186611>\
**Category:** 日本語による質問・議論はこちら\
**Created:** [June 20, 2019, 7:14am UTC](https://discuss.elastic.co/t/aggregate/186611 "2019-06-20T07:14:53Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![harad](https://avatars.discourse-cdn.com/v4/letter/h/e8c25b/32.png) [@harad](https://discuss.elastic.co/u/harad)\
**Post date:** [June 20, 2019, 7:14am UTC](https://discuss.elastic.co/t/aggregate/186611/1 "2019-06-20T07:14:53Z")

</div>

logstash のAggregateフィルタを使って、下記のように、同じIDを目印に、fieldの項目を配列にして一行にしたいと思っています。  
inputはjdbc input pluginを使い、mysqlからデータを取ってきています。

|id\_field|field|  
|100 |AAA |  
|100 |BBB |  
|100 |CCC |  
|200 |DDD |  
|200 |EEE |  
↓  
|id\_field|fields |  
|100 |AAA,BBB,CCC|  
|200 |DDD,EEE |

しかし、以下のように一部まとまらない行があるという状態になっています。

|id\_field|fields |  
|100 |AAA,BBB|  
|100 |CCC |  
|200 |DDD,EEE|

```
filter{
    aggregate {
            task_id => "%{id_field}"
            code => "
            map['id_field'] = event.get('id_field')
            map['fields'] ||= []
            map['fields'] << event.get('field')
            event.cancel()
            "
            push_previous_map_as_event => true
            timeout => 3
    }
}

```

task\_id =\> "%{id\_field}"で指定している限りそうなるはずだと思っているのですが。。  
足りない情報などあればお教え下さい。

---

<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:** [June 20, 2019, 11:21am UTC](https://discuss.elastic.co/t/aggregate/186611/2 "2019-06-20T11:21:26Z")

</div>

> **[Logstash １対多の MySQL データを Elasticsearch にインデックスする（Aggregate Filter 編）](https://medium.com/veltra-engineering/logstash-aggregate-filter-40c27c68483f)**
>
> Aggregate filter で１対多の情報を１つにまとめる

既に確認されているかもしれませんが、こちらで指摘されているような  
ページングやソート順といった内容はクリアされてますでしょうか？

---

<div class="post-metadata">

**Author:** ![yoshioka](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/yoshioka/32/41320_2.png) [@yoshioka](https://discuss.elastic.co/u/yoshioka)\
**Post date:** [June 22, 2019, 2:45pm UTC](https://discuss.elastic.co/t/aggregate/186611/3 "2019-06-22T14:45:28Z")

</div>

aggregate filterは、シングルスレッドでしか正常動作しないので、  
filter処理のworker数を1に設定する必要があります。  
設定されているでしょうか？

設定されていない場合、  
logstsh.ymlでpipeline.workersを1に設定して起動してみてください。  
[https://www.elastic.co/guide/en/logstash/current/logstash-settings-file.html](https://www.elastic.co/guide/en/logstash/current/logstash-settings-file.html)

また、サービス起動ではなく、コマンドライン起動している場合は、  
-w 1 を追加しても同じ設定が可能です。  
[https://www.elastic.co/guide/en/logstash/current/running-logstash-command-line.html](https://www.elastic.co/guide/en/logstash/current/running-logstash-command-line.html)

注意点としては、上記は全てのconfがシングルスレッドで処理されてしまいます。  
動作させるconf毎にworker数を設定する場合は、  
マルチパイプラインで起動して、pipeline.yml側でpipeline毎に  
worker数を設定する必要があります。  
[https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html](https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html)

---

<div class="post-metadata">

**Author:** ![harad](https://avatars.discourse-cdn.com/v4/letter/h/e8c25b/32.png) [@harad](https://discuss.elastic.co/u/harad)\
**Post date:** [June 26, 2019, 6:38am UTC](https://discuss.elastic.co/t/aggregate/186611/4 "2019-06-26T06:38:00Z")

</div>

ご回答ありがとうございます。こちら解決しました。  
2つ原因があり、

1つめは、jsonフィルタを中で使っていたのですが、一部jsonではないデータがありそのフィルタをスキップしていたこと  
2つめはご提示いただいたworker数を1にすることで解決しました。

ありがとうございました。

---

<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:** [July 24, 2019, 6:38am UTC](https://discuss.elastic.co/t/aggregate/186611/5 "2019-07-24T06:38:02Z")

</div>

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