python - 将带有json列的数据从mysql导入到elasticsearch

Question

我在 MySQL 中有一个列，其中一个列中有 json，我必须使用多个键对此列实现搜索。我尝试使用日志存储来使用 Mysql 创建索引。

这是我的日志存储配置。 Info是类型为text和text形式的json对的列

input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://localhost:3306/dbname"
    # The user we wish to execute our statement as
    jdbc_user => "user"
    jdbc_password => "password"
    # The path to our downloaded jdbc driver
    jdbc_driver_library => "/usr/share/java/mysql-connector-java-5.1.38.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    # our query
    statement => "SELECT info FROM organization"
    }
  }
output {
  stdout { codec => json_lines }
  elasticsearch {
  "hosts" => "localhost:9200"
  "index" => "new_index"
  "document_type" => "doc"
  }
}

我尝试创建索引的映射并将其中一个字段设置为嵌套在映射中，但没有任何内容上传到我的索引。 MySQL 对索引的原始更新将我的 json 视为文本，这使得搜索变得更加困难。任何人都有更好的解决方案将 json 列更新为索引，以便我可以从键进行搜索。

输出。

{
  "check_index" : {
    "aliases" : { },
    "mappings" : {
      "doc" : {
        "properties" : {
          "@timestamp" : {
            "type" : "date"
          },
          "@version" : {
            "type" : "text",
            "fields" : {
              "keyword" : {
                "type" : "keyword",
                "ignore_above" : 256
              }
            }
          },
          "info" : {
            "type" : "text",
            "fields" : {
              "keyword" : {
                "type" : "keyword",
                "ignore_above" : 256
              }
            }
          }
        }
      }
    },
    "settings" : {
      "index" : {
        "creation_date" : "1528870439037",
        "number_of_shards" : "5",
        "number_of_replicas" : "1",
        "uuid" : "MkNrBMD8S8GYfDtxRyOFfg",
        "version" : {
          "created" : "6020499"
        },
        "provided_name" : "check_index"
      }
    }
  }
}

有信息是我的 JSON 字符串。在该字段下，我有许多关键值，例如:json 中的地址、名称等，因此我没有为此类字段创建单独的列，而是为其创建了一个 json 并将其添加到该列中。但我无法搜索该 json。

Answer 1

我认为您正在寻找的是 JSON filter。只需在该 JSON 过滤器中添加 JSON 类型的列名称即可。假设如果数据类型 JSON 的列是 info，您的过滤器将如下所示。

filter {
  json {
    source => "info"
    }
}

如果您有多个具有 JSON 数据类型的列，您可以在 filter 中重复您的 json 字典。因此，对于 JSON 列 info，您的最终 Logstash 配置将如下所示。

input {
  jdbc {
      jdbc_connection_string => "jdbc:mysql://localhost:3306/dbname"
      # The user we wish to execute our statement as
      jdbc_user => "user"
      jdbc_password => "password"
      # The path to our downloaded jdbc driver
      jdbc_driver_library => "/usr/share/java/mysql-connector-java-5.1.38.jar"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      # our query
      statement => "SELECT info FROM organization"
  }
} 
filter {
  json {
    source => "info"
    }
}
output {
  elasticsearch {
  "hosts" => "localhost:9200"
  "index" => "new_index"
  "document_type" => "doc"
  }
}