Use filebeat to deal with csv / tsv with header as first line

arr2dict.js

 function convert_csv_to_dict(csv_headers_row, csv_values_row) {
     var json_from_csv =  csv_values_row.reduce(function(result, field, index) {
         result[csv_headers_row[index]] = field;
         return result;
     }, {})
 
     return json_from_csv;
 }
 
 var headers_fn = (function() {
     var csv_headers_row = null;
 
     return function(csv_arr) {
 
     var json_from_csv = null;
 
     if (!csv_headers_row) {
         csv_headers_row = csv_arr;
     } else {
         // combine the csv_headers_row with the values to get a dict
         json_from_csv = convert_csv_to_dict(csv_headers_row, csv_arr)
     }
         return json_from_csv;
     }
 
 })();
 
 function process(event) {
     var csv_arr = event.Get("decoded_csv_arr");
     var json_from_csv = headers_fn(csv_arr);
 
     if (!json_from_csv) {
         event.Cancel();
     }
     event.Put("json_from_csv", json_from_csv);
 }

data.csv

filebeat.yml

# snipits only

 filebeat.inputs: 
 - type: log
   enabled: true
   paths:
     - /your_log_path/*.csv
     
# ------------

 processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
   - add_kubernetes_metadata: ~
   - decode_csv_fields:
       fields:
           message: decoded_csv_arr
       #separator: "\t" # for tsv
       separator: "," # for csv
       ignore_missing: false
       overwrite_keys: true
       trim_leading_space: false
       fail_on_error: true
   #- extract_array:
       #field: decoded_csv_arr
       #mappings:
           #source.ip: 3
   - script:
       lang: javascript
       id: convert_csv_into_json
       file: /your_path/arr2dict.js
   #- timestamp:
       #field: 
   - drop_fields:
       fields: ["decoded_csv_arr"]
 
 output.console:
   pretty: true

Thanks, helpful