# Routing one input to diferent indexes **URL:** \ **Category:** Logstash\ **Created:** [January 9, 2020, 11:35am UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423 "2020-01-09T11:35:36Z")\ **Posts on this page:** 7\ **Page:** 1
**Author:** ![nino](https://avatars.discourse-cdn.com/v4/letter/n/b9bd4f/32.png) [@nino](https://discuss.elastic.co/u/nino)\ **Post date:** [January 9, 2020, 11:35am UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423/1 "2020-01-09T11:35:36Z")
Hello all, i don't know if this can be achieve with logstah. I am just starting with logstash. This is the structure of each file that comes in the folder every day: ``` DATE: FEB 10, 2020 *******section 1******** COLUMN_1 COLUMN_2 COLUMN_3 12 mar 20 3421 15 ene 20 1200 40 mar 20 2102 *******section 2******** COLUMN_1 COLUMN_2 COLUMN_3 17 ene 20 3421 18 feb 20 1200 20 mar 20 2102 TOTAL 107,68 .00 7,830 68,123 SUBTOTAL 40,321 ``` I need to route different parts (matches) of the same file in different indexes (csv files) #### output 1 (routing events/lines to first\_index.csv) @timestamp = Extract the date of the file (top date in the content) TOTAL = 107,68 SUBTOTAL = 40,321 #### output 2 (routing events lines to second\_index.csv) @timestamp = Extract the date of the file (top date in the content) COLUMN\_1 = 17 COLUMN\_2 = ene 20 COLUMN\_3 = 3421 @timestamp = Extract the date of the file (top date in the content) COLUMN\_1 = 18 COLUMN\_2 = feb 20 COLUMN\_3 = 1200 How can i accomplish this? Best regards ---
**Author:** ![nino](https://avatars.discourse-cdn.com/v4/letter/n/b9bd4f/32.png) [@nino](https://discuss.elastic.co/u/nino)\ **Post date:** [January 9, 2020, 6:40pm UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423/3 "2020-01-09T18:40:07Z")
I've just rewrote my question. I hope it is more clear now. Best regards. ---
**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:** [January 9, 2020, 6:43pm UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423/4 "2020-01-09T18:43:25Z")
I would ingest the entire file as a single event and then process it two different ways. With any recent release of logstash you would do that using multiple pipelines and a [forked-path](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html#forked-path-pattern) pattern. However, I will use the old school method. Note the use of literal newlines in patterns. ``` grok { match => { "message" => "^DATE: %{DATA:[@metadata][date]} " } } date { match => ["[@metadata][date]", "MMM dd, yyyy" ] } clone { clones => ["section2"] } if [type] != "section2" { grok { match => { "message" => ["TOTAL\s+(?[0-9,]+)", "SUBTOTAL\s+(?[0-9,]+)" ] } break_on_match => false remove_field => ["message"] } } else { # The (?m) enables multiline matches for the regexp mutate { gsub => ["message", "(?m)^TOTAL.*", "", "message", "(?m).*section 2 \*+\n", ""] } mutate { split => { "message" => " " } } ruby { code => ' events = [] msg = event.get("message") msg.each_index { |x| if x == 0 @column_names = msg[x].split(/\t/) else columns = msg[x].split(/\t\s+/) events << { @column_names[0] => columns[0], @column_names[1] => columns[1], @column_names[2] => columns[2] } end } event.set("events", events) ' remove_field => ["message"] } split { field => "events" } } ``` You would use the same "if [type] !=" to decide which output to write to. If you want the columns at the top level then see [this](https://discuss.elastic.co/t/how-to-dynamically-move-nested-key-value-to-root-level/180006/2). ---
**Author:** ![nino](https://avatars.discourse-cdn.com/v4/letter/n/b9bd4f/32.png) [@nino](https://discuss.elastic.co/u/nino)\ **Post date:** [January 9, 2020, 7:16pm UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423/5 "2020-01-09T19:16:23Z")
Thank you so much Badger, a bit complex for a newbie. Do you need to define [type] in input? ``` clone { clones => ["section2"] } if [type] != "section2" { ``` It's a bit confusing for me the use of [type] and clones this way. What is exactly doing? Best regards ---
**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:** [January 9, 2020, 8:41pm UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423/6 "2020-01-09T20:41:26Z")
The clone filter creates one or more clones of an event. In this example it creates one clone and sets the type to "section2". If you used ``` clone { clones => ["section1", "section2"] } ``` you would get 3 events, and the clone filter would set [type] to "section1" on one of them, and "section2" on the other. The clone filter would not set [type] on the third one. ---
**Author:** ![nino](https://avatars.discourse-cdn.com/v4/letter/n/b9bd4f/32.png) [@nino](https://discuss.elastic.co/u/nino)\ **Post date:** [January 10, 2020, 4:27pm UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423/7 "2020-01-10T16:27:05Z")
Thank you so much Badger. So you apply filters regarding the clone labeled events and route the matches of each labeled events through the corresponding output index. This line it's not mapping the date into @timestamp grok { match =\> { "message" =\> "^DATE: %{DATA:[@metadata][date]} } ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [February 7, 2020, 4:27pm UTC](https://discuss.elastic.co/t/routing-one-input-to-diferent-indexes/214423/8 "2020-02-07T16:27:06Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.