# Multiline vs aggregate **URL:** \ **Category:** Beats\ **Tags:** filebeat\ **Created:** [December 28, 2018, 5:46am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300 "2018-12-28T05:46:21Z")\ **Posts on this page:** 14\ **Page:** 1
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [December 28, 2018, 5:46am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/1 "2018-12-28T05:46:21Z")
I have log file with this structure: > ID,Tag,Value > 10179265,37,21 > 10179265,2001001,15020737 > 10179265,2003001,0 > 10179265,2005000,ARM\_OPERATOR > 10179265,2005002,Int\_OPERATOR > 10160893,37,21 > 10160893,2001001,15054905 > 10160893,2001003,VOICE > 10160893,2003000,0 > 10160893,2003001,0 > 10179296,2004306,20171005130229 > 10179296,2005000, > 10179296,2005001,Roam\_OPERATOR > 10179296,2005002,Int\_OPERATOR > 10179281,37,21 > 10179281,2001001,15038876 the file daily have about 40mil rows I gather the files from production by filebeat I need to merge all lines with same ID into 1 record, all IDs are stored always together in multiple lines the question is ?How to create MULTILINE configuration PATTERN to match lines with same ID? would it be better solution to use agregate filter in logstash? > **[Aggregate filter plugin | Logstash Reference \[8.11\] | Elastic](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-example1)** with aggreagate filter I concern of assumption to set workers to 1 . thank you ---
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [December 28, 2018, 12:34pm UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/2 "2018-12-28T12:34:49Z")
when I wrote this question I found pattern 37,21 I could use to identify start of multiline so this works now > multiline.pattern: 37,21$ > multiline.negate: true > multiline.match: after I receive > "message" =\> "111222,37,21\n111222,2001001,15037925\n111222,2001003,VOICE\n111222,2003000,0" now the issue is logstash part > filter { > csv { > columns =\> ["ID","Tag","Value"] > }} but how to handle \n new lines? ---
**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\ **Post date:** [January 2, 2019, 1:04pm UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/3 "2019-01-02T13:04:11Z")
Looking at your sample `27,21` doesn't look like a good pattern: e.g.; ```auto 10160893,37,21 ... 10179296,2004306,20171005130229 10179296,2005000, 10179296,2005001,Roam_OPERATOR 10179296,2005002,Int_OPERATOR ``` This doesn't look like a smimple multiline problem. Plus, you want to see all tags/values for an ID. At first I'd parse each line into an event of type `{"id": ..., "tag": ..., "value": ...}`. TBH I think that's good enough. By searching for an ID I can immediately see each tag/value. By searching for a tag I can see each ID having said tag. This is great for building visualisations as is. If you still want to combine events, such that `tag` and `value` become arrays you will have to use the aggregate filter after dissect/grok (e.g. combine until ID changes). An event will become `{"id": ..., "tags": [...], "values": [...]}`. ---
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [January 7, 2019, 4:05pm UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/4 "2019-01-07T16:05:38Z")
thank you @stefffe great answer My concern is related to storage consumption by metadata. Daily file has 30Mil rows 1 id is represented by 5-20 rows. No agregation means Each row contains lots of metadata like source, version, timestamp, beat.hostname, beat.name, ID, tags, type, host and payload is two numbers {37=21} Do you think this is good approach to keep it as is and not aggregate it ? The reason for aggregation was to save storage make it faster. There are fields which are not important and I need to index only some of them. But I do not want to remove all fields, I may need then in future. My idea was : from this > 10179296,2004306,20171005130229 > 10179296,2005000, > 10179296,2005001,Roam\_OPERATOR > 10179296,2005002,Int\_OPERATOR to get something like: > message="10179296|2004306=20171005130229|2005000=Test|2005001=Roam\_OPERATOR|2005002=Int\_OPERATOR" and parse only selected fields. for instance > kv { > allow\_duplicate\_values =\> false > field\_split =\> "|" > include\_keys =\> ["2001003","2004305","2005000","2005001","2005002","2001003","2004305","2005000","2005001","2005002"] > } this will give me (different row, but just for reference that there is lot of fields which would normally cost several lines : > "ID" =\> "543567454321", > "2004305" =\> "A", > "2005000" =\> "POST24", > "2005001" =\> "ET", > "2005002" =\> "IntPOST24", > "2001003" =\> "DATA", > "message" =\> "37=21|2001001=83367495|2001003=DATA|2003000=0|2003001=1|2003002=0|2003003=2|2003004=4|2003005=4|2003006=5|2003007=5|2003008=4|2003009=3|2003010=3|2003011=3|2003012=3|2003013=2|2003014=2|2003015=1|2003017=0|2004305=A|2004306=20190223164230|2005000=POST24|2005001=ET|2005002=IntPOST24|2006505=1812|2006506=75000000|2006507=600000000", > "offset" =\> 60557684, > "@timestamp" =\> 2019-01-07T15:38:10.516Z, Thank you for any advice ---
**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\ **Post date:** [January 7, 2019, 8:18pm UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/5 "2019-01-07T20:18:05Z")
Regarding storage, you should do some testing if it really is an issue or some form of premature optimization. I guess the less cardinality a field has, the better it will compress with [doc values](https://www.elastic.co/guide/en/elasticsearch/reference/current/doc-values.html), which is kind of a way to store documents in columnar fashion. One can also [disable `_source`](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-source-field.html), so to reduce disk usage as well. I don't know much about your data or how the data will be consumed which makes it hard for me to make a recommendation. But the way you describe it kinda makes sense as well. I guess it also depends on how you want to query the data. Given the pattern you described I'd use `dissect` + `ruby` filter with a post-aggregate filter one. e.g. `10179296,2004306,20171005130229` dissect: `{"id": 10179296, "key": 2004306, "value": 20171005130229}` ruby: `{"id": 10179296, "2004306": 20171005130229}` aggregate: `{"id": 10179296, "2004306": 20171005130229, "2005000": "Test", ...}` While filebeat has dissect support, the 'key'-transformation + aggregate need to be done in Logstash. ---
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [January 18, 2019, 5:30am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/6 "2019-01-18T05:30:51Z")
Thank you, I plan to test logstash aggregate, but meanwhile I did comparison of single vs multiline option with filebeat. Single ID has 20 rows in average. It means inserting 20rows vs 1 row (using aggregation multiline in filebeat). the result was: | option | rows | Size | Index Time | | --- | --- | --- | --- | | Single | 3,8M | 1,42GB | index\_time: "6.8m", | | Multi | 74,3M | 11,9GB | index\_time": "1.8h", | **SINGLE Line Option** > $ cat 230DEV\_SINGLE.conf > input { > beats { > port =\> 5000 > } > } > > filter { > csv { > columns =\> ["ID","Key","Value"] > skip\_header =\> true > } > } > > output { > elasticsearch { ... **MULTILINE Option** > $ cat 230DEV\_MULTI.conf > input { > beats { > port =\> 5000 > } > } > > filter { > grok { > match =\> {"message" =\> "(?(120|121)[0-9]{9})"} > } > #modify message to remove ID, and create Tag=Value (KV type) > mutate { > gsub =\> [ > "message", "(120|121)[0-9]{9}", "", > "message", "\n,", "|", > "message", "^,", "", > "message", ",", "=" > ] > } > #parse message as KV index only interesting keys > kv { > allow\_duplicate\_values =\> false > field\_split =\> "|" > include\_keys =\> ["2001003","2004305","2005000","2005001","2005002"] > } > > } > > output { > elasticsearch { ... I plan to test aggregation in Logstash as you recommended to see the results. Thank you ---
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [February 5, 2019, 7:26am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/7 "2019-02-05T07:26:09Z")
Hi Steffens, please can you help me how to write ruby code from suggested dissect: `{"id": 10179296, "key": 2004306, "value": 20171005130229}` to transfrom value of key to key and value to value ruby: `{"id": 10179296, "2004306": 20171005130229}` thank you ---
**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\ **Post date:** [February 5, 2019, 4:13pm UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/8 "2019-02-05T16:13:31Z")
See [ruby filter docs](https://www.elastic.co/guide/en/logstash/current/plugins-filters-ruby.html#plugins-filters-ruby-remove_field) and [Event API docs](https://www.elastic.co/guide/en/logstash/current/event-api.html). One could define the ruby filter like this (without testing): ```auto filter { ... ruby { code => "event.set(event.get('key'), event.get('value'))" remove_field => ["key", "value"] } ... } ``` The `remove_field` setting removes the original `key` and `value` fields from the event, if the `code` itself did not fail. ---
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [February 6, 2019, 5:24am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/9 "2019-02-06T05:24:13Z")
Excellent thank you. It was first time I used ruby code. Ruby API reference doc , was the piece I couldnt find. Now it is clear. ---
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [February 7, 2019, 5:41am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/10 "2019-02-07T05:41:12Z")
Finally I incorporated the ruby part into aggregation. Thanks @steffens for help. Index time compared to single line insert vs multiline (filebeat) is ``` * {logstash aggregation} aggreg_test rows:3804206 size:1.7gb indextime: 1,8h inserttime: 3h 20min * {all lines no aggreg } single_test rows:74326466 size:23.8gb indextime: 12,4m inserttime: 2h 45min * {multiline filebeat} multi_test rows:3802915 size:2.8gb indextime: 6,8m inserttime: 14min ``` aggregation provides the most (all is indexed) and least storage however it makes logstash busy the longest. more than 3 hours compared to multiline option 14minut to insert the same file into elasticsearch. Is the aggregation I did correct? Do you have any help why is it so slow? **Aggregation:** filter { dissect { mapping =\> {"message" =\> "%{ID},%{key},%{value}"} } ``` aggregate { task_id => "%{ID}" code => " map['ID'] = event.get('ID') map['profile'] ||= [] map['profile'] << {event.get('key') => event.get('value')} event.cancel() " push_previous_map_as_event => true timeout => 0 #remove_field => ["key", "value"] } ``` } this aggregation provides the result: ``` @timestamp February 6th 2019, 15:42:01.095 t @version 1 t ID 3201235446654 t _id ea5Cw2gBEQ455fveKMOJ t _index aggreg_test # _score - t _type doc ? profile { "37": "21" }, { "2001001": "83793503" }, { "2003000": "0" }, { "2003001": "1" }, { "2003002": "0" }, { "2003003": "2" }, { "2003004": "4" }, { "2003005": "4" }, { "2003006": "5" }, { "2003007": "5" }, { "2003008": "4" }, { "2003009": "3" }, { "2003010": "3" }, { "2003011": "3" }, { "2003012": "3" }, { "2003013": "2" }, { "2003014": "2" }, { "2003015": "1" }, { "2003017": "0" }, { "2004305": "A" }, { "2004306": "20190212154008" }, { "2005000": "Na_Feature" }, { "2005001": "ET" }, { "2005002": "IntNa_Feature" }, { "2006505": "1812" }, { "2006506": "72000000" }, { "2006507": "600000000" }, { "2020151": "0" }, { "2020342": "-2147483647" } ``` ---
**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\ **Post date:** [February 7, 2019, 2:51pm UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/11 "2019-02-07T14:51:30Z")
I'd say the aggregate filter looks fine. Unfortunately I don't have that much experience with this filter to reason about CPU/memory overhead. The aggregated events are somewhat more complex as well. This means: more overhead in serializing, deserializing, indexing. Maybe you should also measure the throughput without ES. For example: ```auto output { stdout { codec => dots } } ``` This will print a dot per event (no encoding). Running LS with `bin/logstash | pv -War >/dev/null` will print the event throughput. How long does it take to process the files this way? ---
**Author:** ![Petr.Simik](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/petr.simik/32/38082_2.png) [@Petr.Simik](https://discuss.elastic.co/u/Petr.Simik)\ **Post date:** [February 12, 2019, 4:08pm UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/12 "2019-02-12T16:08:12Z")
@steffens Thank you so much for all help. the last issue I have is filebeat -\> logstash & singleworker configuration My idea is to create 2 pipelines in Logstash; 1st for all existing logs and 2nd for logs require aggregation (single worker) > logstash]$ less pipelines.yml > > - pipeline.id: main > path.config: "/etc/logstash/conf.d/\*.conf" > > - pipeline.id: singleworker > path.config: "/etc/logstash/singleworker.d/\*.conf" > pipeline.workers: 1 but I need to configure filebeats acordingly > filebeat.prospectors: > > - input\_type: log > paths: > > - input\_type: log > paths: > > output.logstash: > hosts: ["10.0.0.38:5000", "10.0.0.39:5000", "10.0.0.40:5000"] > loadbalance: true > > " HERE I NEED TO ADD IF LOG NEEDS AGGREGATION SEND IT TO DIFFERENT PIPELINE > > - input\_type: log > paths: > - /forAGGREGATION/_.csv > tags: ["singleworker"] > close\_eof: true > clean\_removed: true > enabled: true > close\__: true > harvester\_limit: 5 > > output.logstash: > hosts: ["10.0.0.38:5001", "10.0.0.39:5001", "10.0.0.40:5001"] > loadbalance: true Please what is best practice to configure filebeat -\> logstash in my case filebeat collects n + m logs. m logs needs aggregation. Logstash recieves x +y logs from multiple filebeats and y logs needs aggregation. is my Idea above with 2 pipelines and 1 filebeat correct? how to write if statement in filebeat? we have latest logstash 6.6.0 but filebeat is 5.4 (compiled on Spark/Solaris) thank you ---
**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\ **Post date:** [February 14, 2019, 11:08am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/13 "2019-02-14T11:08:17Z")
You will have to do the event routing in Logstash using the `pipeline` output. See: [https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html) Assuming each pipeline has it's own ES (output) like: ```auto +----------------------------+ | +-----------+ +--------+ | +----------> |LS (single)| |ES (out)| +---------+ | | +-----------+ +--------+ | | +--+ +----+--+ +----------------------------+ v |FB+---->+LS (in)| +-++ +--+ +----+--+ |ES| | +----------------------------+ +-++ | | +----------+ +--------+ | ^ +--------->+ |LS (multi)| |ES (out)| +---------+ | +----------+ +--------+ | +----------------------------+ ``` This approach has 2 potential disadvantages. These very well depend on actual logs and performance of the pipelines: - The ES output in Logstash uses the bulk API, so to reduce overhead when publishing events. If batches are very small you loose potential throughput. - The 2 pipelines are indirectly coupled. Meaning: if one is slow and creates back-pressure, then the back-pressure will slow down the `LS(in)` phase and Filebeat itself. This results in the second pipeline being slowed down as well -\> There will be potentially no performance improvement just by creating 2 separate pipelines this way. When introducing 2 pipelines you want to decouple them, such that they can operate independently. This requires you to duplicate some work or have a potential very very large buffer both pipelines can draw events from independently (still with filtering). This can be solved by either introducing kafka or running 2 filebeat instances (which is somewhat simpler). E.g. by running 2 filebeat instances you will have 2 file readers, each operating at it's own pace. The actual pace will be dictated by ES and the back-pressure generated by the Logstash pipeline itself: ```auto +-----------+ +-----------+ |FB (single)+-------->+LS (single)+---------+ +-----------+ +-----------+ | +v-+ |ES| ++-+ +----------+ +----------+ ^ |FB (multi)+--------->+LS (multi)+----------+ +----------+ +----------+ ``` Drawbacks: - requires separate ports for each pipeline - requires 2 filebeat instances - requires additional filtering in Logstash or Filebeat to send/process only single/multi-line events. Yet you will have completely separate pipelines, allowing the single-line events to outpace the multiline ones. Again, I'd recommend to test/benchmark the 2 architectures to see which one better fits your needs: - Is there really a measurable performance difference - You really want single/multiline events to become visible at a different pace? Not sure if this would suite your data/use-case, but one could also have the single-line event pipeline insert place-holder events (use 'create' action in Elasticsearch output, so to not overwrite existing events) and have the multiline-pipeline overwrite the place-holder events later (use `index` action in Elasticsearch output). ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [March 14, 2019, 11:08am UTC](https://discuss.elastic.co/t/multiline-vs-aggregate/162300/14 "2019-03-14T11:08:23Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.