Custom CSV parser for fluent

config.json

{
    "_default": ["p1","p2", "p3"],
    "event_name123": ["p1", "p2", "p3", "event_name"],
    "event_name321": ["p3", "p2", "p1", "event_name"]
}

fluent.conf

<match events.**>
  @type file
  path /out/csv/files/path
 <format>
    @type starkcsv
    key_name event_name
    format_rules /path/to/config.json
  </format>
</match>

formatter_starkcsv.rb

# put this plugin to the "/etc/td-agent/plugin/formatter_starkcsv.rb"

require 'fluent/plugin/formatter'
require 'csv'
require 'json'

module Fluent
  module Plugin
    class StarkCsvFormatter < Formatter
      Plugin.register_formatter('starkcsv', self)

      config_param :delimiter, default: ',' do |val|
        ['\t', 'TAB'].include?(val) ? "\t" : val
      end
      config_param :force_quotes, :bool, default: true
      config_param :add_newline, :bool, default: true
      config_param :key_name, :string, default: ''
      config_param :format_rules, :string, default: ''

      def configure(conf)
        super

        file = File.read(@format_rules)
        @csv_rules = JSON.parse(file)

        @generate_opts = {col_sep: @delimiter, force_quotes: @force_quotes}
      end

      def format(tag, time, record)
        event_name = record[@key_name]
        fields = @csv_rules[event_name] || @csv_rules['_default']
        row = fields.map do |key|
          record[key]
        end
        line = CSV.generate_line(row, @generate_opts)
        line.chomp! unless @add_newline
        line
      end
    end
  end
end