使用logstash读取MySQL数据传输到es，并且@timestamp字段采用MySQL中的字段时间--建议采用这个

MySQL中数据样式

ES中数据样式

input {

  jdbc {

    jdbc_connection_string => "jdbc:mysql://192.168.0.145:3306/db_example?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC"

    jdbc_user => "root"

    jdbc_password => "root"

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

    jdbc_driver_library => ""

    jdbc_paging_enabled => true

    tracking_column => "unix_ts_in_secs"

    use_column_value => true

    tracking_column_type => "numeric"

    schedule => "*/5 * * * * *"

    statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC"

  }

}

filter {

    # 拆分时间字段获取字符串时间

    dissect{

        mapping => {

            "modification_time" => "%{date}T%{second}.%{?string}"

        }

    }

    # 把字符串时间赋值给新变量，移除字符串时间

    mutate {

       replace => ["modification_time_2","%{[date]} %{[second]}"]

       remove_field => ["date", "second"]

    }

    # 把时间字段的值赋值给@timestamp字段

    ruby {

        code => "event.set('@timestamp',event.get('modification_time'))"

    }

    # 把数据表中的id值赋值给es中的_id,移除无关的字段

    mutate {

        copy => { "id" => "[@metadata][_id]"}

        remove_field => ["id", "@version", "unix_ts_in_secs"]

    }

}

output {

    elasticsearch {

        hosts => ["192.168.75.21:9200"]

        index => "es_table_idx"

        document_id => "%{[@metadata][_id]}"

        user => "elastic"

        password => "GmSjOkL8Pz8IwKJfWgLT"

    }

}

注意：若@timestamp不想使用数据表中的时间字段值，则可以使用系统时间值，但是要使用东八区的系统时间，不是UTC时间

filter {

    # 拆分时间字段获取字符串时间

    dissect{

        mapping => {

            "modification_time" => "%{date}T%{second}.%{?string}"

        }

    }

    # 把字符串时间赋值给新变量，移除字符串时间

    mutate {

       replace => ["modification_time_2","%{[date]} %{[second]}"]

       remove_field => ["date", "second"]

    }

    # 把东八区时间赋值给@timestamp字段

    ruby {

        code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"

    }

    ruby {

        code => "event.set('@timestamp',event.get('timestamp'))"

    }

    # 把数据表中的id值赋值给es中的_id,移除无关的字段

    mutate {

        copy => { "id" => "[@metadata][_id]"}

        remove_field => ["id", "@version", "unix_ts_in_secs"]

    }

}