, action_result: false", :backtrace=>nil}
[INFO] 2023-07-24 02:11:47.370 [LogStash::Runner] runner - Logstash shut down.
```
Logstash aggregate code:
```auto
input {
file {
path => [
"/data/sample_orders.csv",
"/data/sample_users.csv"
]
start_position => "beginning"
sincedb_path => "/dev/null"
codec => plain {
charset => "UTF-8"
}
}
}
filter {
if [source] =~ /sample_orders\.csv/ {
csv {
separator => ","
skip_header => true
columns => ["id", "product", "quantity"]
}
mutate {
convert => {
"id" => "integer"
"quantity" => "integer"
}
}
aggregate {
task_id => "%{id}"
code => "
map['id'] ||= event.get('id')
map['product'] ||= event.get('product')
map['quantity'] ||= event.get('quantity')
event.cancel()
"
push_map_as_event_on_timeout => true
timeout_task_id_field => "id"
timeout => 5
}
}
if [source] =~ /sample_users\.csv/ {
csv {
separator => ","
skip_header => true
columns => ["id", "name", "email"]
}
mutate {
convert => {
"id" => "integer"
}
}
aggregate {
task_id => "%{id}"
code => "
map['name'] ||= event.get('name')
map['email'] ||= event.get('email')
event.cancel()
"
push_previous_map_as_event => true
timeout_task_id_field => "id"
timeout => 5
}
}
ruby {
code => "
event.set('id', event.get('id')) if event.get('id')
event.set('product', event.get('product')) if event.get('product')
event.set('quantity', event.get('quantity')) if event.get('quantity')
event.set('name', event.get('name')) if event.get('name')
event.set('email', event.get('email')) if event.get('email')
event.remove('source')
"
}
}
output {
#elasticsearch {
#hosts => ["localhost:9200"] # Update with the appropriate Elasticsearch host and port
#index => "combined_data_nw" # Replace with the desired index name
#document_id => "%{id}" # Use the 'id' field as the document ID in Elasticsearch
#}
stdout {
codec => rubydebug
}
}
```
---
### Author:  [@coezdemir](https://discuss.elastic.co/u/coezdemir)
#### Post date: [July 24, 2023, 7:46am UTC](https://discuss.elastic.co/t/data-relation-using-logstash-conf-file/338925/4 "2023-07-24T07:46:48Z")
> [@Vinod\_Kumar2](#):
>
> `For task_id pattern '%{id}', there are more than one filter which defines timeout options. All timeout options have to be defined in only one aggregate filter per task_id pattern. Timeout options are : timeout, inactivity_timeout, timeout_code, push_map_as_event_on_timeout, push_previous_map_as_event, timeout_timestamp_field, timeout_task_id_field, timeout_tags>`
It pretty much tells you what is wrong tho.
---
### Author:  [@Vinod\_Kumar2](https://discuss.elastic.co/u/Vinod_Kumar2)
#### Post date: [July 27, 2023, 6:17am UTC](https://discuss.elastic.co/t/data-relation-using-logstash-conf-file/338925/5 "2023-07-27T06:17:26Z")
Hi @Badger
I have removed the timestamp declaration from the code .. it is not combined the data from 2 files. please find the out below
conf file
```auto
input {
file {
path => [
"/data/sample_orders.csv",
"/data/sample_users.csv"
]
start_position => "beginning"
sincedb_path => "/dev/null"
codec => plain {
charset => "UTF-8"
}
}
}
filter {
if [source] =~ /sample_orders\.csv/ {
csv {
separator => ","
skip_header => true
columns => ["id", "product", "quantity"]
}
mutate {
convert => {
"id" => "integer"
"quantity" => "integer"
}
}
aggregate {
task_id => "%{id}"
code => "
map['id'] ||= event.get('id')
map['product'] ||= event.get('product')
map['quantity'] ||= event.get('quantity')
event.cancel()
"
}
}
if [source] =~ /sample_users\.csv/ {
csv {
separator => ","
skip_header => true
columns => ["id", "name", "email"]
}
mutate {
convert => {
"id" => "integer"
}
}
aggregate {
task_id => "%{id}"
code => "
map['name'] ||= event.get('name')
map['email'] ||= event.get('email')
event.cancel()
"
}
}
ruby {
code => "
event.set('id', event.get('id')) if event.get('id')
event.set('product', event.get('product')) if event.get('product')
event.set('quantity', event.get('quantity')) if event.get('quantity')
event.set('name', event.get('name')) if event.get('name')
event.set('email', event.get('email')) if event.get('email')
event.remove('source')
"
}
}
output {
#elasticsearch {
#hosts => ["localhost:9200"] # Update with the appropriate Elasticsearch host and port
#index => "combined_data_nw" # Replace with the desired index name
#document_id => "%{id}" # Use the 'id' field as the document ID in Elasticsearch
#}
stdout {
codec => rubydebug
}
}
```
output:
```auto
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "id,name,email",
"@timestamp" => 2023-07-27T06:08:01.279Z,
"path" => "/data/sample_users.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "id,product,quantity",
"@timestamp" => 2023-07-27T06:08:01.672Z,
"path" => "/data/sample_orders.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "2,Apple,5",
"@timestamp" => 2023-07-27T06:08:01.685Z,
"path" => "/data/sample_orders.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "1,Banana,10",
"@timestamp" => 2023-07-27T06:08:01.686Z,
"path" => "/data/sample_orders.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "3,Orange,3",
"@timestamp" => 2023-07-27T06:08:01.686Z,
"path" => "/data/sample_orders.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "",
"@timestamp" => 2023-07-27T06:08:01.686Z,
"path" => "/data/sample_orders.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "1,John Doe,johndoe@example.com",
"@timestamp" => 2023-07-27T06:08:01.421Z,
"path" => "/data/sample_users.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "2,Jane Smith,janesmith@example.com",
"@timestamp" => 2023-07-27T06:08:01.424Z,
"path" => "/data/sample_users.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "3,Robert Johnson,robertjohnson@example.com",
"@timestamp" => 2023-07-27T06:08:01.424Z,
"path" => "/data/sample_users.csv"
}
{
"host" => "arb_test_vm1",
"@version" => "1",
"message" => "",
"@timestamp" => 2023-07-27T06:08:01.425Z,
"path" => "/data/sample_users.csv"
}
```
Thanks
Vinod
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [July 28, 2023, 12:12am UTC](https://discuss.elastic.co/t/data-relation-using-logstash-conf-file/338925/6 "2023-07-28T00:12:56Z")
Try
```
if [@metadata][path] =~ /sample_users.csv$/ { csv { columns => ["id", "name", "email"] } }
if [@metadata][path] =~ /sample_orders.csv$/ { csv { skip_header => true columns => ["id", "product", "quantity"] } }
aggregate {
task_id => "%{id}"
code => '
map["product"] ||= event.get("product")
map["quantity"] ||= event.get("quantity")
map["name"] ||= event.get("name")
map["email"] ||= event.get("email")
event.cancel
'
push_map_as_event_on_timeout => true
timeout_task_id_field => "id"
timeout => 5
}
```
which will produce events like
```
{
"@timestamp" => 2023-07-28T00:11:20.475338206Z,
"email" => "robertjohnson@example.com",
"quantity" => "3",
"@version" => "1",
"product" => "Orange",
"id" => "3",
"name" => "Robert Johnson"
}
```
---
### Author:  [@Vinod\_Kumar2](https://discuss.elastic.co/u/Vinod_Kumar2)
#### Post date: [July 28, 2023, 4:20am UTC](https://discuss.elastic.co/t/data-relation-using-logstash-conf-file/338925/7 "2023-07-28T04:20:37Z")
Hi @Badger
Super.. It works like a charm.. Thanks a lot
Could you please suggest if we have any samples to learn such kind of scenario's
Thanks
Vinod
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [August 25, 2023, 4:21am UTC](https://discuss.elastic.co/t/data-relation-using-logstash-conf-file/338925/8 "2023-08-25T04:21:01Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.