如何正确使用 NiFi ExtractGrok

Question

我正在使用：NiFi v1.8.0 和 Logstash v7.1.1

我的任务是将我们所有的 Logstash 配置转移到 NiFi。我试图了解 NiFi ExtractGrok 的工作原理，但我找不到任何示例。这打算如何使用？以及如何使用这个 grok 处理器设置 NiFi 属性？当我指的是示例时，我指的是实际示例，这些示例向您展示了之前和之后的情况，以便人们可以了解发生了什么。我已经阅读了 NiFi ExtractGrok 文档，但它非常有限，并且似乎假设您了解它的工作原理。

这是我能找到的唯一例子：How to fetch multiline with ExtractGrok processor in ApacheNifi?

Answer 1

根据您的说法，您需要的处理器是ConvertRecord 而不是ExtractGrok。 ExtractGrok 只会将某些字段提取到 FlowFile 属性或内容中。

如果您想将您的日志文件格式化为一种可行的格式（如 JSON，如果您想将这些文件发送到 ElasticSearch），那么您可以使用GrokReader 作为Record Reader 和Record Writer 作为JsonRecordSetWriter。

然后，您将在RecordReader 和RecordWriter 中配置您的Schema Text（或使用架构注册表）作为您的架构，并将Grok Expression 设置为您在GrokReader 中的grok 表达式。

例如：

我的日志消息记录如下：

2019-12-09 07:59:59,136 this is the first log message
2019-12-09 09:59:59,136 this is the first log message with a stack trace: org.springframework.boot.actuate.jdbc.DataSourceHealthIndicator - DataSource health check failed
org.springframework.jdbc.CannotGetJdbcConnectionException: Failed to obtain JDBC Connection; nested exception is org.apache.commons.dbcp.SQLNestedException: Cannot create PoolableConnectionFactory (Communications link failure
The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server.)
    at org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:81)......

所以，我的想法是： %{TIMESTAMP_ISO8601:timestamp}\s+%{GREEDYDATA:log_message}

我的架构是：

{
  "name": "MyClass",
  "type": "record",
  "namespace": "com.acme.avro",
  "fields": [
    {
      "name": "timestamp",
      "type": "string"
    },
    {
      "name": "log_message",
      "type": "string"
    },
    {
      "name": "stackTrace",
      "type": "string"
    }
  ]
}

注意我添加到架构中的stackTrace 字段。 GrokReader 自动将堆栈跟踪映射到它们自己的字段中。所以如果你也想映射它，你必须添加stackTrace 字段。然后，如果需要，您可以使用 Jolt 将其放入 log_message 字段中。 这个ConvertRecord 的输出是：

[ {
  "timestamp" : "2019-12-09 07:59:59,136",
  "log_message" : "this is the first log message",
  "stackTrace" : null
}, {
  "timestamp" : "2019-12-09 09:59:59,136",
  "log_message" : "this is the first log message with a stack trace: org.springframework.boot.actuate.jdbc.DataSourceHealthIndicator - DataSource health check failed",
  "stackTrace" : "org.springframework.jdbc.CannotGetJdbcConnectionException: Failed to obtain JDBC Connection; nested exception is org.apache.commons.dbcp.SQLNestedException: Cannot create PoolableConnectionFactory (Communications link failure
The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server.)\nat org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:81)......"
} ]