使用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"]
    }

}