Logstash|exception=>\"LogStash::ConfigurationError\",:message=>\"应为 [ \\\\t\\\\r\\\\n], \\\"#\\\" 之一, \\\"=>\\\" 在第 22 行第 20 列

Question

最近在学习logstash时遇到了一个小白问题，尝试将数据从Kafka传输到Bigquery

我正在使用 -

logstash-8.2.3/bin/logstash -f config/logstash-sample.conf

配置文件

input {
  kafka {
        bootstrap_servers => ["addcn102:9092,addcn103:9092,addcn104:9092"]
        group_id => "logstash_kagka"
        auto_offset_reset => "earliest"
        consumer_threads => 3
        topics => ["houstonbq"]
        codec => "json"
  }
}

output {
  google_bigquery {
        project_id => "newcar8891"
        dataset => "logstash"
        json_key_file => "/home/shurui/bin/newcar8891-013ef3777de7.json"
        error_directory => "/opt/module/bqerror"
        flush_interval_secs => "30"
        csv_schema >= "message:STRING"
  }
}  

这是错误日志

Using bundled JDK: /opt/module/logstash-8.2.3/jdk
OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 and will likely be removed in a future release.
Sending Logstash logs to /opt/module/logstash-8.2.3/logs which is now configured via log4j2.properties
[2022-10-21T17:19:07,041][INFO ][logstash.runner          ] Log4j configuration path used is: /opt/module/logstash-8.2.3/config/log4j2.properties
[2022-10-21T17:19:07,051][WARN ][logstash.runner          ] The use of JAVA_HOME has been deprecated. Logstash 8.0 and later ignores JAVA_HOME and uses the bundled JDK. Running Logstash with the bundled JDK is recommended. The bundled JDK has been verified to work with each specific version of Logstash, and generally provides best performance and reliability. If you have compelling reasons for using your own JDK (organizational-specific compliance requirements, for example), you can configure LS_JAVA_HOME to use that version instead.
[2022-10-21T17:19:07,052][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"8.2.3", "jruby.version"=>"jruby 9.2.20.1 (2.5.8) 2021-11-30 2a2962fbd1 OpenJDK 64-Bit Server VM 11.0.15+10 on 11.0.15+10 +indy +jit [linux-x86_64]"}
[2022-10-21T17:19:07,054][INFO ][logstash.runner          ] JVM bootstrap flags: [-Xms1g, -Xmx1g, -XX:+UseConcMarkSweepGC, -XX:CMSInitiatingOccupancyFraction=75, -XX:+UseCMSInitiatingOccupancyOnly, -Djava.awt.headless=true, -Dfile.encoding=UTF-8, -Djruby.compile.invokedynamic=true, -Djruby.jit.threshold=0, -XX:+HeapDumpOnOutOfMemoryError, -Djava.security.egd=file:/dev/urandom, -Dlog4j2.isThreadContextMapInheritable=true, -Djruby.regexp.interruptible=true, -Djdk.io.File.enableADS=true, --add-opens=java.base/java.security=ALL-UNNAMED, --add-opens=java.base/java.io=ALL-UNNAMED, --add-opens=java.base/java.nio.channels=ALL-UNNAMED, --add-opens=java.base/sun.nio.ch=ALL-UNNAMED, --add-opens=java.management/sun.management=ALL-UNNAMED]
[2022-10-21T17:19:07,349][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2022-10-21T17:19:08,125][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600, :ssl_enabled=>false}
[2022-10-21T17:19:08,401][ERROR][logstash.agent           ] Failed to execute action {:action=>LogStash::PipelineAction::Create/pipeline_id:main,
:exception=>"LogStash::ConfigurationError", 
:message=>"Expected one of [ \\t\\r\\n], \"#\", \"=>\" at line 19, column 13 (byte 420) 
after output 
{\n  google_bigquery {\n\t
project_id => \"newcar8891\"\n\t
dataset => \"logstash\"\n\t
csv_schema ",:backtrace=>["/opt/module/logstash-8.2.3/logstash-core/lib/logstash/compiler.rb:32:in `compile_imperative'", 
"org/logstash/execution/AbstractPipelineExt.java:189:in `initialize'",
"org/logstash/execution/JavaBasePipelineExt.java:72:in `initialize'",
"/opt/module/logstash-8.2.3/logstash-core/lib/logstash/java_pipeline.rb:48:in `initialize'",
"/opt/module/logstash-8.2.3/logstash-core/lib/logstash/pipeline_action/create.rb:50:in `execute'", 
"/opt/module/logstash-8.2.3/logstash-core/lib/logstash/agent.rb:381:in `block in converge_state'"]}
[2022-10-21T17:19:08,471][INFO ][logstash.runner          ] Logstash shut down.

这是我的卡夫卡数据

{
    "@timestamp" => 2022-10-21T09:24:51.281353Z,
       "message" => "help me \b",
          "tags" => [
        [0] "_jsonparsefailure"
    ],
      "@version" => "1"
}

我想将消息的内容发送到 bigquery

Answer 1

从错误消息可以理解为您的输出插件配置中存在语法错误。在配置中尝试以下输出插件。

output {
  google_bigquery {
        project_id => "newcar8891"
        dataset => "logstash"
        csv_schema => "message:STRING"
        json_key_file => "/home/shurui/bin/newcar8891-013ef3777de7.json"
        error_directory => "/opt/module/bqerror"
        flush_interval_secs => "30"
  }
}