在 Logstash 中过滤 jdbc 数据答案

【问题标题】：Filter jdbc data in Logstash在 Logstash 中过滤 jdbc 数据
【发布时间】：2019-10-02 08:16:55
【问题描述】：

在我的数据库中，我有以下格式的数据：

但是在ElasticSearch 中，我想推送关于项目类型的数据。因此，ElasticSearch 中的每条记录都会列出每个项目类型的所有项目名称及其值。

像这样：

{
  "_index": "daily_needs",
  "_type": "id",
  "_id": "10",
  "_source": {
    "item_type: "10",
    "fruits": "20",
    "veggies": "32",
    "butter": "11",
  }
}

{
  "_index": "daily_needs",
  "_type": "id",
  "_id": "11",
  "_source": {
    "item_type: "11",
    "hair gel": "50",
    "shampoo": "35",
  }
}

{
  "_index": "daily_needs",
  "_type": "id",
  "_id": "12",
  "_source": {
    "item_type: "12",
    "tape": "9",
    "10mm screw": "7",
    "blinker fluid": "78",
  }
}

我可以在Logstash 中实现这一点吗？

我是 Logstash 的新手，但据我了解，它可以在 filter 中完成。但我不确定要使用哪个过滤器，或者我是否必须为此创建一个自定义过滤器。

当前配置示例：

input {
  jdbc {
    jdbc_driver_library => "ojdbc6.jar"
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    jdbc_connection_string => "myjdbc-configs"
    jdbc_user => "dbuser"
    jdbc_password => "dbpasswd"
    schedule => "* * * * *"
    statement => "SELECT * from item_table"
  }
}
filter {
    ## WHAT TO WRITE HERE??
}
output {
    elasticsearch {
        hosts => [ "http://myeshost/" ]
        index => "myindex"
    }
}

请推荐。 谢谢。

【问题讨论】：

请分享目标索引的当前映射。
@apt-get_install_skill，对不起，我没有得到你的问题。截至目前，我没有使用任何过滤器。表的每一行都像在 ES 中一样被推送。我想知道我可以使用哪个过滤器对数据进行分组。
您可以使用aggregate 过滤器来做到这一点。

标签： elasticsearch jdbc filter logstash

【解决方案1】：

您可以使用aggregate filter plugin 实现此目的。我没有在下面测试过，但应该给你一个想法。

 filter {     
      aggregate {
        task_id => "%{item_type}" #
        code => "
          map['Item_type'] = event.get('Item_type')
          map[event.get('Item_Name')] = map[event.get('Item_Value')]
        "
        push_previous_map_as_event => true
        timeout => 3600
        timeout_tags => ['_aggregatetimeout']
      }
      if "aggregated" not in [tags] {
        drop {}
      }
    }

使用聚合过滤器的重要注意事项：

sql 查询必须按 Item_Type 对结果进行排序，这样事件才不会乱序。
sql 查询中的列名应与过滤器map[] 中的列名匹配
您应该只使用一个工作线程进行聚合，否则可能会乱序处理事件并出现意外结果。

【讨论】：