logstash-columns.conf

input {
    file {
        path => ["/usr/share/logstash/inputs/schemacrawler-columns-*.csv"]
        start_position => "beginning"
    }
}

filter {
    csv {
    skip_header => "true"
    separator => ","
        columns => [
            "sclint-dbenv",
            "sclint-dbid",
            "sclint-catalogName",
            "sclint-runid",
            "sclint-schema-name",
            "sclint-table-name",
            "sclint-table-fullname",
            "sclint-table-remarks",
            "sclint-table-type",
            "sclint-column-short-name",
            "sclint-column-name",
            "sclint-column-fullname",
            "sclint-column-ordinal-position",
            "sclint-column-remarks",
            "sclint-column-default-value",
            "sclint-column-size",
            "sclint-column-type-typename",
            "sclint-column-type-fullname",
            "sclint-column-type-name",
            "sclint-column-type-java-sqltype-group"
        ]
    }

    ruby {
        code => 'find = "false"
                 if !event.get("sclint-column-remarks").nil?
                    if event.get("sclint-column-remarks").include? "RGPD"
                        find = "true"
                    end
                 end
                  event.set("sclint-isRGPDColumn", find)'
    }

    mutate {
        convert => {
            "sclint-column-ordinal-position" => "integer"
            "sclint-column-size" => "integer"
            "sclint-isRGPDColumn" => "boolean"
        }
    }
}

output {
   elasticsearch {
     hosts => "http://elasticsearch:9200"
     index => "schemacrawler-column-stats"
   }
stdout {}
}