logstash同步mysql的数据到elasticsearch时重复

在用logstash将mysql数据同步到es的时候出现了数据重复的问题不知道该怎么去重了，其中attachments/tag/types这三个属性都是数组
大家帮忙看一下：


其中logstash启动文件如下：

input {

  jdbc {

    jdbc_driver_library => "/usr/share/logstash-5.6.2/config/mysql-connector-java-5.1.45.jar"

    jdbc_driver_class => "com.mysql.jdbc.Driver"

    jdbc_connection_string => "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8&useSSL=false&autoReconnect=true&createDatabaseIfNotExist=true"

    jdbc_user => "root"

    jdbc_password => "root"

    jdbc_default_timezone => "Asia/Shanghai"

    jdbc_paging_enabled => true

    jdbc_page_size => 100000

    jdbc_fetch_size => 10000

    connection_retry_attempts => 3

    connection_retry_attempts_wait_time => 1

    jdbc_pool_timeout => 5

    lowercase_column_names => true

    record_last_run => true

    schedule => "* * * * *"

    use_column_value => true

    tracking_column => "id"

    statement_filepath => "/usr/share/logstash-5.6.2/config/knowledge_all.sql"

  }

}

filter {

  aggregate {

    task_id => "%{id}"

    code => "

      map['id'] = event.get('id')

      map['title'] = event.get('title')

      map['attachments'] ||= 

      map['attachments'] << {

        'id' => event.get('attachment_id'),

        'filename' => event.get('attachment_filename'),

        'path' => event.get('attachment_path')

      }

      map['types'] ||= 

      map['types'] << {

        'value' => event.get('type_value'),

        'label' => event.get('type_label')

      }

      map['tag'] ||= 

      map['tag'] << {

        'id' => event.get('tag_id'),

        'title' => event.get('tag_title')

      }

      event.cancel()

    "

    push_previous_map_as_event => true

  }



 }



output {

  elasticsearch {

    hosts => ["localhost:9200"]

    index => "test"

    document_type => "knowledge"

    document_id => "%{id}"

  }

}

knowledge_all.sql:
 

SELECT

DISTINCT

  k.id,

  k.title,

  a.id as attachment_id,

  a.filename as attachment_filename,

  a.path as attachment_path,

  t.id          AS type_value,

  t.title       AS type_label,

  ta.id as tag_id,

  ta.title as tag_title

FROM t_knowledge k

  LEFT JOIN t_knowledge_attachment ka ON ka.knowledge_id = k.id

  LEFT JOIN t_attachment a ON ka.attachment_id = a.id

  LEFT JOIN t_knowledge_type_relate tr ON tr.knowledge_id = k.id

  LEFT JOIN t_knowledge_type t ON t.id = tr.type_id

  LEFT JOIN t_knowledge_tag_relate tar ON tar.knowledge_id = k.id

  LEFT JOIN t_knowledge_tag ta ON ta.id = tar.tag_id