关于 ElasticSearch 和 Mysql 数据同步的问题

我在 mysql 中有两张表 ip 和 ipa

ip 表： |ipid | values          |
          1    192.168.168.168
ipa 表： |ipaid| city | ipid
                 HF      1
即 ipa 表的外键 ipid 与 ip 表的 ipid 对应

我现在想把这两张表的数据同步到 ElasticSearch （简称 ES ），为了关联查询，我首先在 ES 中定义了父子映射：

curl -XPUT "http://localhost:9200/ti?pretty" -d '
{
  "mappings": { 
    "ip" : { }, 
    "ipa":{ 
      "_parent": { 
        "type": "ip"}}
}

在此基础上，我通过 logstash 的 Jdbc input plugin 来实现实时的数据同步，关于 jdbc.conf 的编写：

input {
    stdin {
    }
    jdbc {
      # mysql jdbc connection string to our backup databse
      jdbc_connection_string => "jdbc:mysql://192.168.162.1:3306/model_test"
      # the user we wish to excute our statement as
      jdbc_user => "root"
      jdbc_password => "123"
      # the path to our downloaded jdbc driver
      jdbc_driver_library => "mysql-connector-java-5.1.43.jar"
      # the name of the driver class for mysql
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement_filepath => "jdbc.sql"
      schedule => "* * * * *"
      type => "jdbc_ip"
    }
     jdbc {
      # mysql jdbc connection string to our backup databse
       jdbc_connection_string => "jdbc:mysql://192.168.162.1:3306/model_test"
      # the user we wish to excute our statement as
      jdbc_user => "root"
      jdbc_password => "123"
      # the path to our downloaded jdbc driver
      jdbc_driver_library => "mysql-connector-java-5.1.43.jar"
      # the name of the driver class for mysql
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement_filepath => "jdbc2.sql"
      schedule => "* * * * *"
      type => "jdbc_ip_address"
    }
}
filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}     
output {
 stdout {
        codec => json_lines
    }
    if[type] == "jdbc_ip"{
        elasticsearch {
        hosts  => "localhost:9200"
        index => "ti"
        document_type => "ip"
        document_id => "%{id}"
        }
    }
    if[type] == "jdbc_ip_address"{
        elasticsearch {
        hosts  => "localhost:9200"
        index => "ti"
        document_type => "ipa"
        document_id => "%{id}"
        ## document_parent => "%{ip_id}" ##此处错误
        }
    }
}

我上面的配置文件无法实现：将 Mysql 中的 ip 和 ipa 两张表直接同步到我在 ES 中已建好的 ti 索引
我的问题是，如何通过 logstash 将 Mysql 数据的两张表直接同步 ES 中的索引（索引中有两个文档，为父子关系），或者有没有其他的方法达到此目的。
<附加：ElasticSearch （ 5.2.2 ）、logstash(5.5.2)、环境（ centos6.5)>