Migrate mysql nth level data to elasticsearch via logstash (config file)

[saifrehman]

I am migrating mysql data to elasticsearch via logstash. The database is too big and have lots of relations. I already imported the simple tables and table with nested data(single level parent/child) as well.

Problem : I have user, post and comments table. I want all three table into one type as follow.

> {
> "user_id"   : 200,
> "user_name" : "john doe",
> "posts" : [
>     {
>         post_id         : 1,
>         post_title      : "post description",
>         "comments"  : [
>             {
>                 "comment_id" : 1
>                 "comment_text"  : "good post"
>             },
>             {   "comment_id" : 2
>                 "comment_text"  : "awsome post"
>             }
>         ]
>     },
>     {
>         post_id         : 2,
>         post_title      : "post description",
>         "comments"  : [
>             {
>                 "comment_id" : 3
>                 "comment_text"  : "good post"
>             },
>             {   "comment_id" : 4
>                 "comment_text"  : "awsome post"
>             }
>         ]
>     }
> ]
> }

I also want it to be 3 level nested object. if a user upload a pic in comments then that media is a nested object under comments.

[saifrehman]

I have done this by spending two days. here is my config file.

input {
jdbc{
jdbc_validate_connection => true
jdbc_connection_string => "jdbc:mysql://172.17.0.2:3306/dbname"
jdbc_user => "username"
jdbc_password => "password"
jdbc_driver_library => "/home/ilsa/mysql-connector-java-5.1.36-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
statement => "your query that contain joins "
}
}
filter {
aggregate {
task_id => "%{user_id}"
code => "
map['user_id'] = event.get('user_id')
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 => "index_name"
document_type => "_doc"
document_id => "%{user_id}"
hosts => "localhost:9200"
}
}