ruby - 使用Logstash Ruby过滤器解析csv文件

Question

我有一个Elasticsearch索引，用于索引一组文档。

这些文档最初是csv格式的，我希望使用logstash进行解析。

我的问题是我有以下几方面的事情。

field1,field2,field3,xyz,abc

field3类似于123456789，我想使用ruby代码过滤器将其解析为4.56(789)。

我的尝试:

我尝试使用以下logstash.conf的stdin和stdout。

input {
        stdin {
        }
}

filter {
        ruby {
                code => "
                  b = event["message"]
                  string2=""
                  for counter in (3..(num.size-1))
                         if counter == 4
                                string2+= '_'+ num[counter]
                         elsif counter ==  6
                                string2+= '('+num[counter]
                         elsif counter == 8
                                string2+= num[counter]  +')'
                         else
                                string2+= num[counter]
                         end

                  end

                 event["randomcheck"] = string2

                "
        }
}


output {
        stdout {
                codec=>rubydebug
        }
}

我正在使用此语法错误。

我的最终目标是将其与csv文件一起使用，但是首先我尝试使用stdin和stdout进行此操作。

任何帮助将不胜感激。

Answer 1

出现语法错误的原因很可能是因为您在双引号引起来的字符串中未对双引号进行了转义。可以将字符串用单引号引起来，也可以将其保持双引号，但在内部使用单引号。我也不明白该代码应该如何工作。

除此之外，为什么首先要使用 ruby 过滤器？您可以使用csv过滤器进行CSV解析，并可以使用几个标准过滤器将123456789转换为4.56(789)。

filter {
  # Parse the CSV fields and then delete the 'message' field.
  csv {
    remove_field => ["message"]
  }
  # Given an input such as 123456789, extract 4, 56, and 789 into
  # their own fields.
  grok {
    match => [
      "column3",
      "\d{3}(?<intpart>\d)(?<fractionpart>\d{2})(?<parenpart>\d{3})"
    ]
  }
  # Put the extracted fields together into a single field again,
  # then delete the temporary fields.
  mutate {
    replace => ["column3", "%{intpart}.%{fractionpart}(%{parenpart})"]
    remove_field => ["intpart", "factionpart", "parenpart"]
  }
}

在上面的示例中，临时字段的名称实际上是错误的，因为我不知道它们代表什么。另外，根据输入的外观，您可能需要调整grok表达式。目前，它假设输入的是九位数。