ELK實戰（最詳細）

一、什么是ELK

ELK是三個產品的簡稱：ElasticSearch(簡稱ES) 、Logstash 、Kibana 。其中：

ElasticSearch：是一個開源分布式搜索引擎
Logstash ：是一個數據收集引擎，支持日志搜集、分析、過濾，支持大量數據獲取。其自帶輸入（input）、過濾語法（grok）、輸出（output）三部分，可將數據輸出到ES
Kibana：為 Elasticsearch 提供了分析和 Web 可視化界面

二、如何搭建ELK

各個版本：

ElasticSearch ，Logstash ，Kibana 的版本都為 7.14.0 , JDK：11

1、搭建 ElasticSearch 和 Kibana

參考我之前的博客：

Springboot中使用Elasticsearch（部署+使用+講解最完整）_spring boot elasticsearch-CSDN博客https://blog.csdn.net/qq_73440769/article/details/141477177?spm=1001.2014.3001.5501

2、搭建Logstash

1.查看自己的es版本

docker images

2.拉取鏡像

這個步驟有點慢，可能是我的網絡原因

docker pull docker.elastic.co/logstash/logstash:7.14.0

3.上傳mysql的連接jar包

?可以去IDEA里面復制你Maven里面的：

創建文件夾存放

mkdir -p /opt/logstash/jar

mkdir -p /opt/logstash/jar

4.運行一下鏡像獲取配置文件

docker run -d --name=logstash logstash:7.14.0

第一次創建用于復制文件?

5.查看日志

docker logs -f logstash

6.拷貝數據

docker cp logstash:/usr/share/logstash/config /opt/logstash

docker cp logstash:/usr/share/logstash/data /opt/logstash

docker cp logstash:/usr/share/logstash/pipeline /opt/logstash

7.給文件夾賦權

cd /opt/logstash

chmod -R 777 ./config ./data ./pipeline

8.刪除容器
?

docker rm -f logstash

9.重新啟動容器

docker run -d \--name=logstash \--restart=always \-p 5044:5044 \-v /opt/logstash/data:/usr/share/logstash/data \-v /opt/logstash/jar/mysql-connector-java-8.0.25.jar:/usr/share/logstash/mysql-connector-java-8.0.25.jar \-v /opt/logstash/config:/usr/share/logstash/config \-v /opt/logstash/pipeline:/usr/share/logstash/pipeline \logstash:7.14.0

10.更新配置文件logstash.conf

input {jdbc {jdbc_driver_library => "/usr/share/logstash/mysql-connector-java-8.0.25.jar"jdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_connection_string => "jdbc:mysql://數據庫IP/quick_pickup"jdbc_user => "數據庫用戶名"jdbc_password => "數據庫密碼"statement => "SELECT id AS id,openid AS openid,quick_user_id AS quickUserId,name AS name,sex AS sex,avatar AS avatar,phone AS phone,follow AS follow,fan AS fan,wallet AS wallet,DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%s') AS createTime,use_time AS useTime,collect_number AS collectNumber,mark_number AS markNumber,brief_introduction AS briefIntroductionFROM user"lowercase_column_names => false # 關閉傳輸字段默認小寫的配置# 開啟分頁jdbc_paging_enabled => truejdbc_page_size => 2000schedule => "*/5 * * * * * UTC"  # 每5秒執行一次}
}output {elasticsearch {hosts  => ["es所在服務器的IP:9200"]index  => "user"                 # Elasticsearch 索引名稱document_id => "%{id}"           # 使用 MySQL 的主鍵 `id` 作為文檔 IDcodec  => "json"}
}

11.修改logstash.yml

12.重啟容器

docker stop logstash

docker start logstash

或者：
?

docker restart logstash

13.再次打印日志查看

docker logs -f logstash

三、提醒

記得打開服務器對應的端口（5044）

四、可能遇到的bug

下面是我之前遇到的問題，最后都解決了，上面配置文件是最新更新后的配置文件

logstash輸出到es的字段都是小寫
時間字段不是我們希望的格式

這些bug可以參考我上面編寫更新后的配置文件，下面給出github一個大佬的一些匯總，可以參考一下：

https://github.com/logstash-plugins/logstash-filter-date/issues/158https://github.com/logstash-plugins/logstash-filter-date/issues/158

#logstash輸入配置
input {#jdbc輸入配置，用來指定mysql中需要同步的數據查詢SQL及同步周期jdbc {type => "jdbc"jdbc_connection_string => "jdbc:mysql://localhost:3306/dh_order?autoReconnect=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false"# 數據庫連接賬號密碼；jdbc_user => "dh_test"jdbc_password => "Y2017dh123"# MySQL依賴包路徑；jdbc_driver_library => "mysql/mysql-connector-java-5.1.49.jar"jdbc_driver_class => "com.mysql.jdbc.Driver"# 數據庫重連嘗試次數connection_retry_attempts => "3"# 判斷數據庫連接是否可用，默認false不開啟jdbc_validate_connection => "true"# 數據庫連接可用校驗超時時間，默認3600Sjdbc_validation_timeout => "3600"# 是否開啟分頁jdbc_paging_enabled => true# 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"# statement => "SELECT * FROM `t_car_order` limit 1"statement => "SELECT id,create_time FROM `t_car_order` limit 1"# 是否將字段名轉換為小寫，默認true（如果有數據序列化、反序列化需求，建議改為false）；# lowercase_column_names => false# Value can be any of: fatal,error,warn,info,debug，默認info；# sql_log_level => warnsql_log_level => debug# 是否記錄上次執行結果，true表示會將上次執行結果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；# record_last_run => true# 需要記錄查詢結果某字段的值時，此字段為true，否則默認tracking_column為timestamp的值；# use_column_value => true# 需要記錄的字段，用于增量同步，需是數據庫字段# tracking_column => "ModifyTime"# Value can be any of: numeric,timestamp，Default value is "numeric"# tracking_column_type => timestamp# record_last_run上次數據存放位置；# last_run_metadata_path => "mysql/last_id.txt"# 是否清除last_run_metadata_path的記錄，需要增量同步時此字段必須為false；# clean_run => false# 設置定時任務間隔  含義：分、時、天、月、年，全部為*默認含義為每分鐘跑一次任務，這里設置為每5分鐘同步一次# schedule => "*/5 * * * * *"# 用來控制增量更新的字段，一般是自增id或者創建、更新時間，注意這里要采用sql語句中select采用的字段別名# tracking_column => "unix_ts_in_secs"# tracking_column 對應字段的類型# tracking_column_type => "numeric"}
}
#logstash輸入數據的字段匹配和數據過濾
# filter {
#   mutate {
#     copy => { "id" => "[@metadata][_id]"}
#     remove_field => ["id", "@version", "unix_ts_in_secs"]
#   }
# }
filter {# date {#   match => ["update_time", "yyyy-MM-dd HH:mm:ss"]#   target => "update_time"# }# date {#   match => ["create_time", "yyyy-MM-dd HH:mm:ss"]#   target => "create_time"# }# mutate {#   convert => { "create_time" => "text" }   # 將create_time字段轉換為字符串類型# }# ruby {#   code => 'event.set("create_time", event.get("create_time").strftime("%Y-%m-%d %H:%M:%S"))'# }#  date {#   match => ["create_time", "yyyy-MM-dd HH:mm:ss"]#   target => "create_time"#   timezone => "Asia/Shanghai" # 你的時區# }mutate {add_field => { "index_date" => "%{create_time}" }}# mutate {#   rename => { "create_time_string" => "index_date" }# }#  date {#   # match => ["index_date", "ISO8601"]#   match => ["index_date", "ISO8601"]#   # target => "index_date"# }# }date {match => ["index_date", "yyyy-MM-dd HH:mm:ss"]# target => "index_date"# target => "index_date"}# mutate {#   add_field => {#     "index_date1" => "%{index_date}"#   }}
#logstash輸出配置
output {# 采用stdout可以將同步數據輸出到控制臺，主要是調試階段使用# stdout { codec =>  json_lines}stdout { codec =>  rubydebug}# 指定輸出到ES的具體索引# elasticsearch {#     index => "rdbms_sync_idx"#     document_id => "%{[@metadata][_id]}"# }elasticsearch {# host => "192.168.1.1"# port => "9200"# 配置ES集群地址# hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]hosts => ["localhost:9200"]# 索引名字，必須小寫# index => "t_car_order-%{+YYYY.MM.dd}"index => "t_car_order_%{index_date}"# index => "t_car_order_@timestamp"# index => "t_car_order3"# 數據唯一索引（建議使用數據庫KeyID）# document_id => "%{KeyId}"document_id => "%{id}"# document_id => "ID"}
}

?關于字段大小寫問題還可以參考這幾篇博客：

Elasticsearch-logstash同步mysql數據字母大小寫問題_es 字段小寫-CSDN博客文章瀏覽閱讀2.5k次。logstash同步mysql數據的時候，sql里面含有的大寫字母，到了ES的時候就會變成小寫，這是因為在jdbc.conf里面沒有添加lowercase_column_names => false"這個屬性，就導致es里面看到的字段名稱全是小寫。最后總結：es是支持大寫字段名稱的，如果想要保留原有的大寫字母，需要在同步配置中加上lowercase_column_names ..._es 字段小寫https://blog.csdn.net/qinyuezhan/article/details/89215215

Logstash將字段名全部轉換為小寫 - 騰訊云開發者社區 - 騰訊云Logstash是一個開源的數據收集引擎，用于將不同來源的數據進行收集、轉換和傳輸。它是Elastic Stack（Elasticsearch、Logstash、Kibana）中的一部分，用于處理和分......https://cloud.tencent.com/developer/information/Logstash%E5%B0%86%E5%AD%97%E6%AE%B5%E5%90%8D%E5%85%A8%E9%83%A8%E8%BD%AC%E6%8D%A2%E4%B8%BA%E5%B0%8F%E5%86%99-salon