# Logstash - how to configure CSV filter for joining 2 CSV files based on a common field (mapping - one to many data) and ingest the data into single index
**URL:** https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994
**Category:** Logstash
**Created:** [January 1, 2021, 12:49pm UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994 "2021-01-01T12:49:48Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@Jenisha\_Ramanathan](https://discuss.elastic.co/u/Jenisha_Ramanathan)
#### Post date: [January 1, 2021, 12:49pm UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994/1 "2021-01-01T12:49:48Z")
Hi Team,
I have 2 CVS files which contains one similar column header say "faculty\_id",
For the rows which has same "faculty\_id" value, I want the below steps to be done
1. Join/combine the data from both the csv files into 1 row, and ingest this row into ElasticSearch.
2. If the left or right part in the join is empty, replace the empty with value 'null'
Find the below column header for CSV files,
**Column Header of CSV1**
faculty\_id,faculty\_name,reporting\_manager
**Column Header of CVS2**
faculty\_id,student\_name,batch,year,percentage
**Sample Input Data**
csv1:
faculty\_id,faculty\_name,reporting\_manager
1,AAA,R1
2,BBB,R2
csv2:
faculty\_id,student\_name,batch,year,percentage
1,STU1,2k20,2020,90
1,STU2,2k20,2020,78
2,STU3,2k20,2020,85
1,STU4,2k20,2020,75
2,STU5,2k20,2020,80
3,STU6,2k20,2020,80
**Sample Result:**
faculty\_id,faculty\_name,reporting\_manager,student\_name,batch,year,percentage
1,AAA,R1,STU1,2k20,2020,90
1,AAA,R1,STU2,2k20,2020,78
2,BBB,R2,STU3,2k20,2020,85
1,AAA,R1,STU4,2k20,2020,75
2,BBB,R2,STU5,2k20,2020,80
3,null,null,STU6,2k20,2020,80
Could anyone please guide me with the Logstash configuration to achieve the above mentioned scenario.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 1, 2021, 4:27pm UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994/2 "2021-01-01T16:27:24Z")
You will need to reformat the csv1 file, but you can do it using a translate filter.
Use a file input to read csv2 and then use the following filters
```
csv { autodetect_column_names => true }
translate {
dictionary_path => "/home/user/t.test/csv1.csv"
field => "faculty_id"
destination => "faculty_name"
}
```
Then you would use a second translate filter that has a file that looks like
```
faculty_id,reporting_manager
1,R1
2,R2
```
Alternately, completely changing the format of csv1
```
faculty_id,JSON
1,"{""faculty_name"": ""AAA"", ""reporting_manager"": ""R1""}"
2,"{""faculty_name"": ""BBB"", ""reporting_manager"": ""R2""}"
```
then use these filters
```
csv { autodetect_column_names => true }
translate {
dictionary_path => "/home/user/csv1.csv"
field => "faculty_id"
destination => "[@metadata][json]"
}
json { source => "[@metadata][json]" }
```
---
### Author:  [@Jenisha\_Ramanathan](https://discuss.elastic.co/u/Jenisha_Ramanathan)
#### Post date: [January 4, 2021, 5:39am UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994/3 "2021-01-04T05:39:35Z")
Hi Badger,
Thank you for the quick response.
Do we need to have two separate input file for csv1. I am stuck in creating the file with json input with id as you mentioned above.
And also please suggest if we can use mutate filter for any kind of conversion in this.
Could you please provide the sample config for this scenario if any.
#csv1-\> "/opt/parquet/test-input1.csv"
#csv2-\>"/opt/parquet/test-input2.csv"
input {
file {
type =\> "file2"
path =\> "/opt/parquet/test-input2.csv"
start\_position =\> "beginning"
sincedb\_path =\> "/dev/null"
}
}
filter {
if [type] == "file2" {
csv {
columns =\> ["...."id"...."]
separator =\> ","
}
}
mutate {
convert =\> {"percentage" =\> "float"}
}
```
translate {
dictionary_path => "/opt/parquet/input1.csv"
field => "id"
destination => "name"
```
}
}
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 4, 2021, 3:34pm UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994/4 "2021-01-04T15:34:27Z")
As I said, you would have to have three csv files. test-input2.csv would remain as is. The other two would be
test1.csv
```
faculty_id,faculty_name
1,AAA
2,BBB
```
test3.csv
```
faculty_id,reporting_manager
1,R1
2,R2
```
Then use two translate filters
```
translate {
dictionary_path => "/some/path/test1.csv"
field => "faculty_id"
destination => "faculty_name"
}
translate {
dictionary_path => "/some/path/test3.csv"
field => "faculty_id"
destination => "reporting_manager"
}
```
---
### Author:  [@Jenisha\_Ramanathan](https://discuss.elastic.co/u/Jenisha_Ramanathan)
#### Post date: [January 6, 2021, 5:35am UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994/5 "2021-01-06T05:35:30Z")
As you suggested, have had three files and used two translate filters but getting the below exception,
[ERROR][logstash.agent] Failed to execute action {:id=\>:sample\_cur\_pipeline, :action\_type=\>LogStash::ConvergeResult::FailedAction, :message=\>"Could not execute action: PipelineAction::Create\, action\_result: false", :backtrace=\>nil}.
Please guide me with this, also find the below configuration used for the same.
input {
file {
path =\> "/opt/parquet/test1.csv"
start\_position =\> "beginning"
sincedb\_path =\> "/dev/null"
}
}
filter {
csv {
columns =\> ["faculty\_id","student\_name","batch","year","percentage"]
separator =\> ","
}
translate {
dictionary\_path =\> "/opt/parquet/test2.csv"
field =\> "faculty\_id"
destination =\> "faculty\_name"
}
translate {
dictionary\_path =\> "/opt/parquet/test3.csv"
field =\> "faculty\_id"
destination =\> "reporting\_manager"
}
}
output {
elasticsearch {
hosts =\> ["[http://aa.bb.cc.dd](http://aa.bb.cc.dd):xxx"]
index =\> "xxx-index"
user =\> "uuu"
password =\> "bbb"
}
}
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 6, 2021, 4:51pm UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994/6 "2021-01-06T16:51:21Z")
Try enabling --log.level debug and see if you get a more informative error message.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [February 3, 2021, 4:51pm UTC](https://discuss.elastic.co/t/logstash-how-to-configure-csv-filter-for-joining-2-csv-files-based-on-a-common-field-mapping-one-to-many-data-and-ingest-the-data-into-single-index/259994/7 "2021-02-03T16:51:23Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.