elasticsearch - Logstash MalformedCSVError:行内引用非法|如何配置logstash conf以读取多列mutiline CSV？

Question

已经通过CSV::MalformedCSVError: Illegal quoting in line 1 with SmarterCSV
和Rescue CSV::MalformedCsvError: Illegal quoting in line n
他们都没有解决我的问题。

我在将CSV conf与Log Stash基本集成时遇到问题:

    input {
      file {
        path => "/softwares/imews_customer_wallet_details.csv"
        start_position => "beginning"
      }
    }
    filter {
      csv {
        columns => ["imiles_wallet_id","creation_date","gcid_cod","total_imiles","is_active"]
        separator => ","
      }
      date {
        match => ["creation_date", "dd-MM-yy"]
        target => "creation_date"
      }
    }
    output {
      elasticsearch {
        action => "index"
        hosts => ["localhost:9200"]
        index => "ime_customers"
      }
    }

错误堆栈在循环上打印以下行:

    [2020-03-02T13:36:51,968][WARN ][logstash.filters.csv     ][another_test] Error parsing csv {:field=>"message", :source=>"{\"level\":\"DEBUG\",\"timeUTC\":\"Mon Feb 24 08:05:20 UTC 2020\",\"timeUnix\":1582531520800,\"appID\":\"TEST\",\"hub\":\"CI\",\"environment\":\"NONE\",\"host\":\"Yashpal3150153\",\"logMessage\":\"http-outgoing-1 << Content-Type: application/x-amz-json-1.0\",\"threadID\":\"SimpleAsyncTaskExecutor-1\",\"cdll_version\":\"6.5.0\"}\r", :exception=>#<CSV::MalformedCSVError: Illegal quoting in line 1.>}

我想念什么，这是怎么了？

更新:

CSV:

    125,IME,"14-11-19","14-11-19",IME,"14-11-19",gcid3,100,TRUE
    124,IME,"14-11-19","14-11-19",IME,"14-11-19",gcid2,100,TRUE
    123,IME,"14-11-19","14-11-19",IME,"14-11-19",gcid1,252,TRUE

更新:
管道:

     - pipeline.id: another_test
       queue.type: persisted
       path.config: "/softwares/logstash-7.6.0/config/*.conf"

配置包含:

我提到的conf是: cust.conf
另外，我将cust.conf更新为:

    input {
      file {
        path => "D:/softwares/imews_customer_wallet_details.csv"
        start_position => "beginning"
      }
    }
    filter {
      csv {
        columns => ["imiles_wallet_id","creation_date","gcid_cod","total_imiles","is_active"]
        separator => ","
        codec => plain
        convert => {
          "imiles_wallet_id" => "integer"
          "is_active" => "boolean"
          "creation_date" => "date"
          "gcid_cod" => "string"
          "total_imiles" => "integer"
        } 
      }
    }
    output {
      elasticsearch {
        action => "index"
        hosts => ["localhost:9200"]
        index => "ime_customers"
      }
    }

现在错误是:

    LogStash::Error: Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<another_test>`
              create at org/logstash/execution/ConvergeResultExt.java:109
                 add at org/logstash/execution/ConvergeResultExt.java:37
      converge_state at D:/softwares/logstash-7.6.0/logstash-core/lib/logstash/agent.rb:339
    [2020-03-04T18:20:05,345][ERROR][logstash.agent           ] An exception happened when converging configuration {:exception=>LogStash::Error, :message=>"Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<another_test>`", :backtrace=>["org/logstash/execution/ConvergeResultExt.java:109:in `create'", "org/logstash/execution/ConvergeResultExt.java:37:in `add'", "D:/softwares/logstash-7.6.0/logstash-core/lib/logstash/agent.rb:339:in `block in converge_state'"]}
    [2020-03-04T18:20:05,397][FATAL][logstash.runner          ] An unexpected error occurred! {:error=>#<LogStash::Error: Don't know how to handle `Java::JavaLang::IllegalStateException` for `PipelineAction::Create<another_test>`>, :backtrace=>["org/logstash/execution/ConvergeResultExt.java:109:in `create'", "org/logstash/execution/ConvergeResultExt.java:37:in `add'", "D:/softwares/logstash-7.6.0/logstash-core/lib/logstash/agent.rb:339:in `block in converge_state'"]}
    ```

Answer 1

通过使用管道定义/softwares/logstash-7.6.0/config/*.conf启动Logstash，您可能会有许多相互干扰的配置文件(即，几个不同的输入，过滤器和输出)。所有这些配置文件都合并为一个配置。

阅读this article，以了解如何最好地组织Logstash管道。
同时，您可以像这样从命令行简单地启动Logstash:

bin/logstash -f /softwares/logstash-7.6.0/config/cust.conf