本文作者：羅海鵬，叩丁狼高級(jí)講師烫映。原創(chuàng)文章瞧预，轉(zhuǎn)載請(qǐng)注明出處。
案例一：收集Nginx訪問(wèn)日志數(shù)據(jù)
Nginx是企業(yè)中常用的http服務(wù)器榨惰，也有人稱它為反向代理服務(wù)器和負(fù)債均衡服務(wù)器拜英，憑借著性能強(qiáng)大和功能完善在整個(gè)互聯(lián)網(wǎng)行業(yè)中大量使用，所以收集Nginx服務(wù)器的訪問(wèn)日志數(shù)據(jù)琅催，并且轉(zhuǎn)存到elasticsearch中居凶，就是一個(gè)很常見的需求了，然后再根據(jù)elasticsearch強(qiáng)大的搜索和聚合能力統(tǒng)計(jì)出我們的應(yīng)用的訪問(wèn)量藤抡、訪問(wèn)用戶所在地侠碧、訪問(wèn)了什么功能、訪問(wèn)時(shí)間和使用了什么客戶端訪問(wèn)的等等信息缠黍。
創(chuàng)建配置文件
首先弄兜，我們創(chuàng)建一個(gè)Logstash的配置文件，文件名隨意瓷式，我們起名為：logstash.conf替饿，在該文件中編輯input、filter和output這3個(gè)組件的配置贸典。
然后视卢，在啟動(dòng)Logstash實(shí)例的時(shí)候，使用-f參數(shù)的方式啟動(dòng)廊驼，參數(shù)值為該配置文件的存放路徑据过，比如：./logstash -f /soft/logstash_conf/logstash.conf
input配置
input {
file {
path => "/usr/local/nginx/logs/access.log"
type => "nginx-access"
start_position => "beginning"
}
}
我們使用了一個(gè)file的輸入插件，讀取的文件是nginx的訪問(wèn)日志access.log妒挎，存放在：/usr/local/nginx/logs/目錄中绳锅。nginx訪問(wèn)日志默認(rèn)格式為：
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent" "$http_x_forwarded_for"';
如果需要自定義訪問(wèn)日志的格式，只需要在nginx.conf配置文件中修改以上內(nèi)容酝掩。
通過(guò)以上格式輸出的日志會(huì)寫到/usr/local/nginx/logs/access.log文件中鳞芙，內(nèi)容為：
192.168.85.1 - - [16/Sep/2018:00:42:38 +0800] "GET /catalog/getChildCatalog?catalogId=1 HTTP/1.1" 200 3249 "http://mgrsite.shop.com/" "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36
一旦nginx的訪問(wèn)日志有新的內(nèi)容更新，會(huì)被Logstash監(jiān)控到庸队，并讀取到Logstash中积蜻，然后Logstash為該數(shù)據(jù)流創(chuàng)建一個(gè)Event對(duì)象，我們可以在output組件中設(shè)置一個(gè)標(biāo)準(zhǔn)輸出：stdout彻消，在顯示器打印Event對(duì)象，打印結(jié)果為：
{
"message" => "192.168.85.1 - - [16/Sep/2018:00:42:38 +0800] "GET /catalog/getChildCatalog?catalogId=1 HTTP/1.1" 200 3249 "http://mgrsite.shop.com/" "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36",
"@version" => "1",
"@timestamp" => 2018-08-13T17:32:01.122Z,
"host" => "localhost.localdomain"
}
上面的打印結(jié)果就是一個(gè)Event對(duì)象宙拉，在該Event對(duì)象中的message字段封裝了一條數(shù)據(jù)流的數(shù)據(jù)（注意：在上一節(jié)中宾尚，我們提到了file輸入插件默認(rèn)是以換行符作為一條數(shù)據(jù)流的，可以通過(guò)配置來(lái)設(shè)置數(shù)據(jù)流的分隔符）
filter配置
filter {
grok {
match => {
"message" => "%{IP:remote_ip} \- \- \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status} %{NUMBER:bytes} \"%{URI:domain}\" \"%{GREEDYDATA:user_agent}"
}
remove_field => ["message","@timestamp","@version"]
}
date {
match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z"]
target => "timestamp"
}
if [remote_ip] !~ "^127\.|^192\.168\.|^172\.1[6-9]\.|^172\.2[0-9]\.|^172\.3[01]\.|^10\." {
geoip {
source => "remote_ip"
}
}
}
在Logstash過(guò)濾的組件中，我們使用了3個(gè)過(guò)濾插件煌贴，分別是grok御板、date和geoip，該3個(gè)插件從上外下進(jìn)行3層的數(shù)據(jù)過(guò)濾和轉(zhuǎn)化牛郑。
我們按順序一個(gè)一個(gè)來(lái)看看這些過(guò)濾插件的配置：
（1）怠肋、grok插件
在grok插件中，我們拿到了Event對(duì)象中的message字段淹朋，該message字段封裝了nginx訪問(wèn)日志文件最新追加的一條數(shù)據(jù)笙各，并且使用match匹配這條日志數(shù)據(jù)，該匹配的正則為（grok正則匹配語(yǔ)法在上一章提到础芍，如果忘了可以回顧上一章）：
%{IP:remote_ip} \- \- \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status} %{NUMBER:bytes} \"%{URI:domain}\" \"%{GREEDYDATA:user_agent}
那通過(guò)了該grok插件過(guò)濾后杈抢，就會(huì)在原來(lái)的Event對(duì)象中新增remote_ip、timestamp仑性、method惶楼、request、http_version诊杆、status歼捐、bytes、domain和user_agent這幾個(gè)字段晨汹，字段的值是由message的值匹配過(guò)來(lái)的窥岩。與此同時(shí)，我們還刪除了一些不需要的字段宰缤，比如：”message”,”@timestamp”,”@version”颂翼。
（2）、date插件
我們?cè)趃rok過(guò)濾插件匹配到message中的時(shí)間后慨灭，使用一個(gè)新的字段存儲(chǔ)朦乏，該字段為timestamp，但是該時(shí)間的格式是不太直觀氧骤，是dd/MMM/YYYY:HH:mm:ss Z這樣的格式呻疹，打印出來(lái)的數(shù)據(jù)就是16/Sep/2018:00:42:38 +0800。我們需要把這種格式的時(shí)間轉(zhuǎn)成比較直觀的筹陵，比如：YYYY-MM-dd HH:mm:ss,所以刽锤，我們?cè)赿ate插件中配置了match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z"]，意思是匹配到timestamp字段中的時(shí)間格式為dd/MMM/YYYY:HH:mm:ss Z朦佩，到時(shí)候并思，date插件會(huì)把該匹配到的時(shí)間轉(zhuǎn)成YYYY-MM-dd HH:mm:ss格式，并把轉(zhuǎn)換好的時(shí)機(jī)重新覆蓋到timestamp字段中语稠，也就是我們target => "timestamp"這行配置的體現(xiàn)宋彼，如果沒有這個(gè)配置弄砍，那么轉(zhuǎn)換后的時(shí)間默認(rèn)覆蓋到”@timestamp”，但是該字段已經(jīng)被我們移除了输涕。
（3）音婶、geoip插件
我們?cè)趃rok過(guò)濾插件匹配到message中的訪問(wèn)IP后，使用一個(gè)新的字段存儲(chǔ)莱坎，該字段為remote_ip衣式，此時(shí)，我們?nèi)绻行枰ㄟ^(guò)IP地址獲取歸屬地的話檐什，就需要使用到geoip插件了碴卧，但是我們?cè)谑褂迷摬寮坝昧艘粋€(gè)邏輯判斷，排除了127厢汹、192螟深、168和172這些內(nèi)網(wǎng)IP，如果是以這些內(nèi)網(wǎng)IP訪問(wèn)的話烫葬，就不使用geoip插件了（該判斷的語(yǔ)法在上一章有提界弧，如果忘了可以回顧上一章），因?yàn)閮?nèi)網(wǎng)IP是么有歸屬地的搭综。如果獲取到了歸屬地等信息后垢箕，geoip會(huì)在Event對(duì)象中新增一個(gè)字段，字段名叫：geoip兑巾，然后該字段對(duì)應(yīng)的值又是一個(gè)對(duì)象条获，那么在默認(rèn)情況下，geoip對(duì)象包含十多個(gè)字段蒋歌，具體可回顧上一章geoip插件的詳解帅掘。
綜合上述的3個(gè)過(guò)濾插件后，最終得到的Event為：
{
"request" => "/property/get/1",
"method" => "GET",
"type" => "nginx-access",
"status" => "200",
"http_version" => "1.1",
"user_agent" => "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36\" \"-\"",
"path" => "/usr/local/nginx/logs/access.log",
"timestamp" => 2018-09-20T14:17:48.000Z,
"host" => "bogon",
"bytes" => "503",
"remote_ip" => "120.77.150.83",
"domain" => "http://mgrsite.shop.com/",
"geoip" => {
"ip" => "120.77.150.83",
"continent_code" => "AS",
"city_name" => "Hangzhou",
"region_name" => "Zhejiang",
"longitude" => 120.1614,
"region_code" => "33",
"country_code2" => "CN",
"country_code3" => "CN",
"latitude" => 30.2936,
"timezone" => "Asia/Shanghai",
"country_name" => "China",
"location" => {
"lat" => 30.2936,
"lon" => 120.1614
}
}
}
output配置
output {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "logstash-%{type}"
document_type => "%{type}"
}
}
以上配置是在output組件加上一個(gè)elasticsearch 插件堂油，把一個(gè)Event對(duì)象寫入到elasticsearch中修档，完成nginx訪問(wèn)日志的收集。
最后府框，我們把上述的配置放到一起查看吱窝，這樣更直觀的看到整個(gè)access_log.conf配置文件的完整內(nèi)容：
input {
file {
path => "/usr/local/nginx/logs/access.log"
type => "nginx-access"
start_position => "beginning"
}
}
filter {
grok {
match => {
"message" => "%{IP:remote_ip} \- \- \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status} %{NUMBER:bytes} \"%{URI:domain}\" \"%{GREEDYDATA:user_agent}"
}
remove_field => ["message","@timestamp","@version"]
}
date {
match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z"]
target => "timestamp"
}
if [remote_ip] !~ "^127\.|^192\.168\.|^172\.1[6-9]\.|^172\.2[0-9]\.|^172\.3[01]\.|^10\." {
geoip {
source => "remote_ip"
}
}
}
output {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "logstash-%{type}"
document_type => "%{type}"
}
stdout {
}
}
驗(yàn)證數(shù)據(jù)
我們可以先借助head界面來(lái)查看elasticsearch轉(zhuǎn)存過(guò)來(lái)的nginx訪問(wèn)日志數(shù)據(jù)（后面需要把head升級(jí)為kibana），能看到以下截圖的數(shù)據(jù)迫靖，說(shuō)明配置沒有問(wèn)題院峡，否則再檢查一下配置文件。
案例二：MySQL數(shù)據(jù)全量&增量導(dǎo)入到ES
在企業(yè)中往往會(huì)有這樣的情況系宜，原本某張表的數(shù)據(jù)不需要做全文搜索照激，現(xiàn)在需要做了，但是我們知道關(guān)系型數(shù)據(jù)庫(kù)不適合做全文搜索蜈首，那么我們就需要想辦法把關(guān)系數(shù)據(jù)庫(kù)中的數(shù)據(jù)轉(zhuǎn)移到elasticsearch中做全文搜索了实抡。
還有一種情況欠母，某張關(guān)系型數(shù)據(jù)庫(kù)的表欢策，原本只需要做簡(jiǎn)單的like模糊查詢就可以了吆寨，但是隨著該表的數(shù)據(jù)量日益增加，那么在大數(shù)據(jù)量下踩寇，like查詢效率非常低啄清，影響用戶使用體驗(yàn)，更嚴(yán)重的是俺孙，由于處理一個(gè)請(qǐng)求效率低辣卒，那可能導(dǎo)致大量的請(qǐng)求阻塞堆積，服務(wù)器壓力過(guò)大睛榄，最終導(dǎo)致應(yīng)用宕機(jī)荣茫。那為了避免這樣的問(wèn)題發(fā)生，我們就需要想辦法把大數(shù)據(jù)表轉(zhuǎn)移到elasticsearch场靴，根據(jù)elasticsearch特點(diǎn)啡莉，它的搜索速度極快，并且不會(huì)因?yàn)閿?shù)據(jù)量的增加而導(dǎo)致搜索效率的下降旨剥。
創(chuàng)建配置文件
上面的案例我們已經(jīng)創(chuàng)建了一個(gè)Logstash的配置文件了：logstash.conf咧欣，所以我們就直接使用logstash.conf文件，不再創(chuàng)建一個(gè)新的配置文件來(lái)做這個(gè)案例轨帜，這樣就方便我們學(xué)習(xí)和測(cè)試階段魄咕。但是大家需要知道，如果這樣的話蚌父，但時(shí)候我們選擇logstash.conf來(lái)啟動(dòng)Logstash實(shí)例時(shí)哮兰，該實(shí)例就會(huì)做兩件事情，收集nginx訪問(wèn)日志和導(dǎo)入mysql數(shù)據(jù)苟弛，這樣對(duì)該Logstash的壓力是比較大的喝滞，所以，在實(shí)際應(yīng)用中嗡午，更建議一個(gè)Logstash實(shí)例就做一件事情囤躁，比如創(chuàng)建兩個(gè)配置文件：nginx_access.conf和mysql.conf，分別代表兩個(gè)Logstash實(shí)例荔睹。
input配置
input {
file {
path => "/usr/local/nginx/logs/access.log"
type => "nginx-access"
start_position => "beginning"
}
jdbc{
type => "product"
jdbc_driver_library => "/soft/logstash_config/mysql-connector-java-5.1.46.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://192.168.85.1:3306/wolfcode_shop_goods"
jdbc_user => "root"
jdbc_password => "root"
statement => "select * from product where id > :sql_last_value"
jdbc_paging_enabled => true
jdbc_page_size => 2
use_column_value => true
tracking_column => "id"
last_run_metadata_path => "/soft/logstash_config/mysql_record_info"
schedule => "* * * * *"
}
}
我們?cè)谠瓉?lái)的input組件上狸演，在新增了一個(gè)jdbc插件，該插件使用jdbc接口讀取所有實(shí)現(xiàn)了jdbc標(biāo)準(zhǔn)的關(guān)系型數(shù)據(jù)庫(kù)的數(shù)據(jù)僻他，而我們這次的案例使用的關(guān)系型數(shù)據(jù)庫(kù)是mysql宵距。
以上配置就是jdbc輸入插件的配置，使用該配置和elasticsearch輸出插件配合使用吨拗，能完成mysql全量和增量的數(shù)據(jù)導(dǎo)入满哪。其中jdbc_paging_enabled婿斥、use_column_value、tracking_column和last_run_metadata_path這4個(gè)屬性配置是實(shí)現(xiàn)增量導(dǎo)入的關(guān)鍵哨鸭，如果沒有配置這4個(gè)屬性民宿，則只能實(shí)現(xiàn)全量導(dǎo)入。
mysql數(shù)據(jù)全量&增量導(dǎo)入的過(guò)程
我們來(lái)看看logstash是如果實(shí)現(xiàn)mysql數(shù)據(jù)全量&增量導(dǎo)入的：
1像鸡、首先活鹰，因?yàn)槲覀兣渲昧?code>use_column_value => true和
tracking_column => "id"，意思是logstash在程序內(nèi)存中可以使用查詢出來(lái)的product表中id這個(gè)列只估。2志群、然后，我們配置
schedule => "* * * * *"代表每分鐘執(zhí)行一次讀取數(shù)據(jù)的任務(wù)蛔钙，而last_run_metadata_path => "/soft/logstash_config/mysql_record_info"這行配置意思是：把上一次讀取product表數(shù)據(jù)時(shí)最后一條數(shù)據(jù)的值锌云，記錄在mysql_record_info文件中，因?yàn)槲覀?code>tracking_column配置的是id這一列吁脱，所以該mysql_record_info文件記錄的就是上一次讀取數(shù)據(jù)時(shí)桑涎，最后一條數(shù)據(jù)的id值。3豫喧、這時(shí)石洗，我們啟動(dòng)Logstash實(shí)例，會(huì)把
mysql_record_info文件記錄的值賦值給程序的sql_last_value變量紧显，那所以讲衫，我們每次讀取數(shù)據(jù)發(fā)送的SQL查詢語(yǔ)句：statement => "select * from product where id > :sql_last_value"，意思就是只查詢上一次記錄的id值之后的數(shù)據(jù)孵班。4涉兽、如果logstash在讀取
mysql_record_info文件的時(shí)候，發(fā)現(xiàn)該文件沒有記錄內(nèi)容篙程，那說(shuō)明之前沒有讀取過(guò)數(shù)據(jù)枷畏，這時(shí)logstash就把sql_last_value變量設(shè)置為0。-
5虱饿、到達(dá)定時(shí)任務(wù)的時(shí)間拥诡，從
sql_last_value變量記錄的值開始讀取數(shù)據(jù)。logstash會(huì)發(fā)送一條count語(yǔ)句統(tǒng)計(jì)product表共有多少條數(shù)據(jù)氮发，然后在根據(jù)我們?cè)O(shè)置的jdbc_page_size => 2（默認(rèn)是10萬(wàn)條）每頁(yè)查詢2條數(shù)據(jù)渴肉，計(jì)算出需要發(fā)送多少條select語(yǔ)句才能把product表剩余的數(shù)據(jù)讀取出來(lái)（因?yàn)?code>sql_last_value變量的值為0，所以這時(shí)相當(dāng)于全量查詢）爽冕，并且把最后一條數(shù)據(jù)的id重新賦值給sql_last_value變量仇祭，同時(shí)，把該值重新寫在mysql_record_info文件中颈畸，供以后重啟Logstash實(shí)例時(shí)使用乌奇，重啟實(shí)例又進(jìn)行第3個(gè)步驟没讲。以下語(yǔ)句是logstash執(zhí)行全量查詢時(shí)發(fā)送的SQL語(yǔ)句：SELECT count(*) AS `count` FROM (select * from product where id > 0) AS `t1` LIMIT 1 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 0 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 2 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 4 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 6 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 8 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 10 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 12 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 14 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 16 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 18 SELECT * FROM (select * from product where id > 0) AS `t1` LIMIT 2 OFFSET 20 -
6、再次到達(dá)定時(shí)任務(wù)的時(shí)間礁苗，又執(zhí)行第5步驟的操作（但不同的是
sql_last_value變量已經(jīng)不為0了爬凑，所以此時(shí)相當(dāng)于增量查詢）。我們新增3條數(shù)據(jù)做測(cè)試寂屏，以下語(yǔ)句是logstash執(zhí)行增量查詢時(shí)發(fā)送的SQL語(yǔ)句：SELECT count(*) AS `count` FROM (select * from product where id > 44) AS `t1` LIMIT 1 SELECT * FROM (select * from product where id > 44) AS `t1` LIMIT 2 OFFSET 0 SELECT * FROM (select * from product where id > 44) AS `t1` LIMIT 2 OFFSET 2 7贰谣、Logstash獲取到j(luò)dbc輸入進(jìn)來(lái)的數(shù)據(jù)后娜搂，配合后面需要配置的elasticsearch輸出插件迁霎，把數(shù)據(jù)寫到elasticsearch中。
Logstash周而復(fù)始的進(jìn)行5,6,7的步驟百宇，實(shí)現(xiàn)mysql數(shù)據(jù)全量&增量導(dǎo)入考廉。在這個(gè)過(guò)程中，我們講解了幾個(gè)關(guān)鍵的jdbc插件屬性配置的作用携御，如果還有其他屬性是不知道它們的作用的昌粤，可以回顧上一章輸入插件：jdbc的詳細(xì)介紹。
filter配置
filter {
if [type] == "nginx-access" {
grok {
match => {
"message" => "%{IP:remote_ip} \- \- \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status} %{NUMBER:bytes} \"%{URI:domain}\" \"%{GREEDYDATA:user_agent}"
}
remove_field => ["message","@timestamp","@version"]
}
date {
match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z"]
target => "timestamp"
}
if [remote_ip] !~ "^127\.|^192\.168\.|^172\.1[6-9]\.|^172\.2[0-9]\.|^172\.3[01]\.|^10\." {
geoip {
source => "remote_ip"
}
}
}
}
由于Logstash通過(guò)jdbc輸入插件啄刹，拿到的數(shù)據(jù)暫時(shí)不需要做什么解析涮坐，所以不需要配置過(guò)濾插件，這里就直接使用邏輯判斷數(shù)據(jù)來(lái)源誓军，如果type是nginx-access袱讹，才進(jìn)入grok和date過(guò)濾插件的處理。
output配置
output {
if [type] == "nginx-access" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "logstash-%{type}"
document_type => "%{type}"
}
}
if [type] == "product" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "mysql-%{type}"
document_type => "%{type}"
document_id => "%{id}"
}
}
stdout {
}
}
最后昵时，我們?cè)谠瓉?lái)的output組件中再增肌一個(gè)elasticsearch輸出插件捷雕，并且使用邏輯判斷數(shù)據(jù)來(lái)源，分別儲(chǔ)存到同一個(gè)elasticsearch搜索服務(wù)器中的不同索引庫(kù)中壹甥，并且type=product的插件中救巷，我們配置了document_id => "%{id}"，意思是使用mysql中查詢出來(lái)的id句柠，作為文檔id
驗(yàn)證數(shù)據(jù)
最后浦译，我們看看elasticsearch中是新增了mysql_porduct索引庫(kù)、product文檔類型和文檔數(shù)據(jù)：
想獲取更多技術(shù)視頻溯职，請(qǐng)前往叩丁狼官網(wǎng)：http://www.wolfcode.cn/openClassWeb_listDetail.html
