使用Logstash把MySQL數據導入到Elasticsearch中

總結：這種適合把已有的MySQL數據導入到Elasticsearch中

有一個csv文件，把里面的數據通過Navicat Premium 軟件導入到數據表中，共有998條數據

文件下載地址：https://files.cnblogs.com/files/sanduzxcvbnm/SalesJan2009.zip

csv文件格式如下：

Logstash 配置
1.下載連接mysql的驅動包，放到指定目錄下
在地址https://dev.mysql.com/downloads/connector/j/下載最新的Connector。下載完這個Connector后，把這個connector存入到Logstash安裝目錄下的如下子目錄中：

logstash-core/lib/jars/

conf文件內容如下：

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_validate_connection => true
       jdbc_driver_library => ""
       jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
       parameters => { "Product_id" => "Product1" }
       statement => "SELECT * FROM salesjan2009 WHERE Product = :Product_id"
    }    
}
 
filter {
	mutate {
	 	rename => {
        	"longitude" => "[location][lon]"
        	"latitude" => "[location][lat]"
    	}
    }
}
 
output {
    stdout {
    
    }
 
   elasticsearch {
        hosts => ["192.168.75.21:9200"]
     	index => "sales" # 指定索引名
     	document_type => "_doc"
        user => "elastic"
        password => "GmSjOkL8Pz8IwKJfWgLT"
    } 
}

說明：
1.jdbc_connection_string => "jdbc:mysql://192.168.0.145:3306/db_example?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC"
連接的數據庫地址，端口號，數據庫名，字符編碼，時區等,db_example為數據庫名
2.
jdbc_user => "root"
jdbc_password => "root"
連接數據庫使用的用戶名和密碼，根據自己的實際情況而定

3.jdbc_driver_library
驅動包路徑，若是在logstash指定目錄下則留空，若不是則需要指定絕對路徑

4.jdbc_driver_class
最新使用的驅動包類

5.parameters
設置一個參數Product_id，其值是Product1

6.statement
sql語句，結合上面的理解，是查詢salesjan2009數據表中條件Product的值是Product_id也即是Product1的數據

7.filter mutate
新增一個字段，重構經緯度參數值結構

運行Logstash來加載我們的MySQL里的數據到Elasticsearch中：

./bin/logstash --debug -f  ./config/conf.d/sales.conf 

可以在Kibana中查看到最新的導入到Elasticsearch中的數據：

注意數據總數，並不是數據表中的全部數據，而是根據查詢條件獲得的部分數據。