如何使用logstash将嵌套的mysql对象索引到elasticsearch中？答案

【问题标题】：How to index nested mysql object into elasticsearch using logstash?如何使用logstash将嵌套的mysql对象索引到elasticsearch中？
【发布时间】：2016-08-09 14:30:17
【问题描述】：

我正在尝试使用 elasticsearch 索引 mysql 数据库。考虑示例映射：

{"blog":
  {"properties": 
    {"id": "string"}
    {"author": "string"}
    {"time_created": }
    {"author_info": 
      {"author_name":}
      {"author_sex":}
    }
    {"posts":
      {"post_author":}
      {"post_time":}
    }
  }
}

我有三个表，它们是 author_info、blog 和 post。如何将这些记录索引到具有嵌套结构的弹性中？我找不到有关它的文件。谢谢

【问题讨论】：

杨锐，你解决了吗？我遇到了同样的问题，我无法将嵌套的 mysql 数据放入具有与您的结构非常相似的 elasticsearch 中。谢谢。

标签： mysql elasticsearch logstash

【解决方案1】：

在logstash 输入的sql 部分中，您可以尝试在elasticsearch 中选择具有您想要的嵌套名称的字段。下面是它的外观的一个小示例。

输入{ jdbc {

statement => "SELECT id as blog.properties.id, author as blog.properties.author,..... from blog inner join properties inner join posts"

} }

【讨论】：

我对弹性搜索不是很熟悉。您能否指定如何格式化 sql 查询以使其结构化？
使用 logstash jdbc 输入的示例编辑答案 - 应该比在原始答案中使用 elasticsearch-jdbc 更接近您所在的位置。
@bbergvt 这不起作用，它在尝试这样嵌套时显示 MySQL 语法错误：select author as blog.properties.author -- 还有其他建议吗？

【解决方案2】：

    input {
    jdbc{
        jdbc_validate_connection => true
        jdbc_connection_string => "jdbc:mysql://172.17.0.2:3306/_db"
        jdbc_user => "root"
        jdbc_password => "admin"
        jdbc_driver_library => "/home/ilsa/mysql-connector-java-5.1.36-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        clean_run => true   
        statement => "SELECT  
            u.id as employee_number, u.email as email, u.username as username, 
            up.id as post_id, up.text_content as content, 
            pc.id as comment_id , pc.user_post_id as comment_post_id, pc.comment as comment_text 
            FROM users u join user_posts up on up.user_id = u.id 
            LEFT JOIN  post_comments pc ON pc.user_post_id = up.id 
            ORDER BY up.id ASC"
    }

}
filter {
    aggregate {
        task_id => "%{employee_number}"
        code => "
            map['employee_number'] = event.get('employee_number')
            map['email'] = event.get('email')
            map['username'] = event.get('username')
            map['posts'] ||= []
            map['posts'] << {

                'post_id' => event.get('post_id'),
                'content' => event.get('content'),
                'comments' => [] << { 
                    'comment_id' => event.get('comment_id'),
                    'comment_post_id' => event.get('comment_post_id'),
                    'comment_text' => event.get('comment_text')
                }

            }
        event.cancel()"
        push_previous_map_as_event => true
        timeout => 30
    }
}
output {
    stdout{ codec => rubydebug }
    elasticsearch{
        action => "index"
        index => "_dev"
        document_type => "_doc"
        document_id => "%{employee_number}"
        hosts => "localhost:9200"
    }

}

【讨论】：