# Cannot combine two indexes to one using logstash **URL:** \ **Category:** Logstash\ **Created:** [May 21, 2019, 8:10pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083 "2019-05-21T20:10:31Z")\ **Posts on this page:** 8\ **Page:** 1
**Author:** ![aldol](https://avatars.discourse-cdn.com/v4/letter/a/9f8e36/32.png) [@aldol](https://discuss.elastic.co/u/aldol)\ **Post date:** [May 21, 2019, 8:10pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/1 "2019-05-21T20:10:32Z")
I have the following two indexes: PUT twitter1/doc/1 { "name" : "Mike", "c\_num" : 1, "tags" : "in1" } PUT twitter2/doc/1 { "resume" : "Mike blah, blah, blah", "c\_num" : 1, "tags" : "in2" } I would like to have a final index combining two. I like to have both "name" and "resume" for each c\_num. Like below: ## twitter4 { "resume" : "Mike blah, blah, blah", "name" : "Mike", "c\_num" : 1, } I used following logstash but for some reason filtering doesn't join them together: input { elasticsearch { hosts =\> "[http://11.222.33.44:9200](http://11.222.33.44:9200)" user =\> "user" password =\> "pwd" index =\> "twitter1" } elasticsearch { hosts =\> "[http://11.222.33.44:9200](http://11.222.33.44:9200)" user =\> "user" password =\> "pwd" index =\> "twitter2" } } filter { aggregate { task\_id =\> "%{c\_num}" ``` code => " if (event.get('tags').include('in1')) map['name'] = event.get('name') else map['resume'] = event.get('resume') end event.cancel(); " inactivity_timeout => 300 #seconds since last event push_map_as_event_on_timeout => true timeout_task_id_field => "c_num" ``` } } output { elasticsearch { hosts =\> "[http://11.222.33.44:9200](http://11.222.33.44:9200)" ssl\_certificate\_verification =\> "false" index =\> "twitter4" document\_id =\> "%{c\_num}" } } --- I am getting the following error: "Aggregate exception occurred {:error=\>#\," and the result is not what I want. I am open for any logstash alternative solution. ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [May 21, 2019, 10:04pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/2 "2019-05-21T22:04:22Z")
You have not made tags an array, it is just a string, so you should do a string equality test, not a .include test ---
**Author:** ![aldol](https://avatars.discourse-cdn.com/v4/letter/a/9f8e36/32.png) [@aldol](https://discuss.elastic.co/u/aldol)\ **Post date:** [May 22, 2019, 12:24pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/3 "2019-05-22T12:24:50Z")
Thanks Badger for your comment. I changed filter as below: filter { aggregate { task\_id =\> "%{c\_num}" ``` code => " if (event.get('tags') != 'in1') map['resume'] = event.get('resume') else map['name'] = event.get('name') end event.cancel(); " inactivity_timeout => 300 #seconds since last event push_map_as_event_on_timeout => true timeout_task_id_field => "c_num" ``` } But I get one record from Twitter1 and one from twitter 2 and they are not combined. "hits" : [ { "\_index" : "twitter4", "\_type" : "doc", "\_id" : "2", "\_score" : 1.0, "\_source" : { "@version" : "1", "@timestamp" : "2019-05-21T19:54:23.850Z", "resume" : "David blah, blah, blah", "c\_num" : 2, "tags" : [ "in2", "\_aggregateexception" ] } }, { "\_index" : "twitter4", "\_type" : "doc", "\_id" : "1", "\_score" : 1.0, "\_source" : { "@version" : "1", "@timestamp" : "2019-05-21T19:54:23.864Z", "name" : "Mike", "c\_num" : 1, "tags" : [ "in1", "\_aggregateexception" ] } ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [May 22, 2019, 2:17pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/4 "2019-05-22T14:17:25Z")
They have different c\_num values, so I would not expect them to be aggregated. Does the aggregate filter log an exception? ---
**Author:** ![aldol](https://avatars.discourse-cdn.com/v4/letter/a/9f8e36/32.png) [@aldol](https://discuss.elastic.co/u/aldol)\ **Post date:** [May 22, 2019, 3:06pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/5 "2019-05-22T15:06:59Z")
Aggregate filter doesn't log an exception. In my example in Twitter1 and Twitter2 have the same c\_num value. The whole idea is to combine two indexes to make one index and have columns from both indexes. One index has a column "name" and other index has column "resume" for the same c\_num. I would like final index twitter4 to have both "name" and "resume" column for the same c\_num. Below should be the final result: twitter4 { "resume" : "Mike blah, blah, blah", "name" : "Mike", "c\_num" : 1, } So I am looking for the correct filter to do this. ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [May 22, 2019, 8:11pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/6 "2019-05-22T20:11:45Z")
> [@aldol](#): > > Aggregate filter doesn't log an exception. It is hard to believe that no exception is logged when the events output include an \_aggregateexception tag. ---
**Author:** ![aldol](https://avatars.discourse-cdn.com/v4/letter/a/9f8e36/32.png) [@aldol](https://discuss.elastic.co/u/aldol)\ **Post date:** [May 23, 2019, 12:44pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/7 "2019-05-23T12:44:33Z")
First thank you for following up. I am not expert in logstash and may be I didn't understand exactly what you are asking for. But I meant when I run the logstash, I don't get error in aggregation. As a matter of fact it says."Aggregate successful". Do I need to look somewhere else to see the exception? Here is the result displays on the screen. [logstash.agent] Successfully started Logstash API endpoint {:port=\>1234} [logstash.instrument.periodicpoller.jvm] collector name {:name=\>"ParNew"} [logstash.instrument.periodicpoller.jvm] collector name {:name=\>"ConcurrentMarkSweep"} [logstash.pipeline] filter received {"event"=\>{"@version"=\>"1", "@timestamp"=\>2019-05-23T12:28:32.970Z, "tags"=\>"in2", "control\_num"=\>2, "resume"=\>"David blah, blah, blah"}} [logstash.pipeline] filter received {"event"=\>{"@timestamp"=\>2019-05-23T12:28:32.985Z, "name"=\>"Mike", "tags"=\>"in1", "control\_num"=\>1, "@version"=\>"1"}} [logstash.pipeline] filter received {"event"=\>{"@timestamp"=\>2019-05-23T12:28:32.970Z, "name"=\>"David", "tags"=\>"in1", "control\_num"=\>2, "@version"=\>"1"}} [logstash.pipeline] filter received {"event"=\>{"@version"=\>"1", "@timestamp"=\>2019-05-23T12:28:32.981Z, "tags"=\>"in2", "control\_num"=\>1, "resume"=\>"Mike blah, blah, blah"}} [logstash.inputs.elasticsearch] Closing {:plugin=\>"LogStash::Inputs::Elasticsearch"} [logstash.filters.aggregate] Aggregate successful filter code execution {:code=\>"\r\n map['tags'] = event.get('tags')\r\n if (event.get('tags') != 'in1')\r\n map['resume'] = event.get('resume')\r\n else\r\n map['name'] = event.get('name')\r\n end\r\n# event.cancel()\r\n "} [logstash.filters.aggregate] Aggregate successful filter code execution {:code=\>"\r\n map['tags'] = event.get('tags')\r\n if (event.get('tags') != 'in1')\r\n map['resume'] = event.get('resume')\r\n else\r\n map['name'] = event.get('name')\r\n end\r\n# event.cancel()\r\n "} [logstash.filters.aggregate] Aggregate successful filter code execution {:code=\>"\r\n map['tags'] = event.get('tags')\r\n if (event.get('tags') != 'in1')\r\n map['resume'] = event.get('resume')\r\n else\r\n map['name'] = event.get('name')\r\n end\r\n# event.cancel()\r\n "} [logstash.filters.aggregate] Aggregate successful filter code execution {:code=\>"\r\n map['tags'] = event.get('tags')\r\n if (event.get('tags') != 'in1')\r\n map['resume'] = event.get('resume')\r\n else\r\n map['name'] = event.get('name')\r\n end\r\n# event.cancel()\r\n "} [logstash.pipeline] output received {"event"=\>{"@timestamp"=\>2019-05-23T12:28:32.970Z, "name"=\>"David", "tags"=\>["in1"], "control\_num"=\>2, "@version"=\>"1"}} [logstash.pipeline] output received {"event"=\>{"@timestamp"=\>2019-05-23T12:28:32.985Z, "name"=\>"Mike", "tags"=\>["in1"], "control\_num"=\>1, "@version"=\>"1"}} [logstash.pipeline] output received {"event"=\>{"@version"=\>"1", "@timestamp"=\>2019-05-23T12:28:32.970Z, "tags"=\>["in2"], "control\_num"=\>2, "resume"=\>"David blah, blah, blah"}} [logstash.pipeline] output received {"event"=\>{"@version"=\>"1", "@timestamp"=\>2019-05-23T12:28:32.981Z, "tags"=\>["in2"], "control\_num"=\>1, "resume"=\>"Mike blah, blah, blah"}} [logstash.pipeline] Pushing flush onto pipeline {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutting down filter/output workers {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Setting shutdown {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Flushing {:plugin=\>#LogStash::FilterDelegator:0x54555fb2} [logstash.filters.aggregate] Aggregate timeout for '%{control\_num}' pattern: 1800 seconds [logstash.pipeline] Flushing {:plugin=\>#LogStash::FilterDelegator:0x54555fb2} [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.pipeline] Flushing {:plugin=\>#LogStash::FilterDelegator:0x54555fb2} [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.filters.aggregate] Aggregate remove\_expired\_maps call with '%{control\_num}' pattern and 2 maps [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.pipeline] Shutdown waiting for worker thread {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.filters.aggregate] Closing {:plugin=\>"LogStash::Filters::Aggregate"} [logstash.filters.aggregate] Aggregate close call {:code=\>"\r\n map['tags'] = event.get('tags')\r\n if (event.get('tags') != 'in1')\r\n map['resume'] = event.get('resume')\r\n else\r\n map['name'] = event.get('name')\r\n end\r\n# event.cancel()\r\n "} [logstash.outputs.elasticsearch] Closing {:plugin=\>"LogStash::Outputs::ElasticSearch"} [logstash.outputs.elasticsearch] Stopping sniffer [logstash.outputs.elasticsearch] Stopping resurrectionist [logstash.outputs.elasticsearch] Waiting for in use manticore connections [logstash.outputs.elasticsearch] Closing adapter #LogStash::Outputs::ElasticSearch::HttpClient::ManticoreAdapter:0x29ab1495 [logstash.pipeline] Pipeline has terminated {:pipeline\_id=\>"main", :thread=\>"#\"} [logstash.instrument.periodicpoller.os] Stopping [logstash.instrument.periodicpoller.jvm] Stopping [logstash.instrument.periodicpoller.persistentqueue] Stopping [logstash.agent] Shutting down all pipelines {:pipelines\_count=\>0} [logstash.agent] Converging pipelines state {:actions\_count=\>0}[INFO][logstash.runner] Logstash shut down. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [June 20, 2019, 12:44pm UTC](https://discuss.elastic.co/t/cannot-combine-two-indexes-to-one-using-logstash/182083/8 "2019-06-20T12:44:42Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.