一. 摘要
Elasticsearch作為大數(shù)據(jù)場景下的搜索和分析的引擎巷送，廣泛的應(yīng)用于實時數(shù)據(jù)的分析場景。本文介紹如何通過logstash_input_jdbc插件將mysql數(shù)據(jù)準(zhǔn)實時的同步于Elasticsearch剃毒。
二. 前期準(zhǔn)備
- 安裝Elasticsearch（本文采用的版本為2.4.6）
- 安裝Mysql
- 安裝Logstash（本文采用的版本為6.1.1）
由于本文主要介紹如何Mysql數(shù)據(jù)向Elasticsearch同步實操步驟筷笨，前期準(zhǔn)備工作請參閱其他文章。
三. 實驗步驟
3.1 安裝logstash_input_jdbc插件
進入logstash bin目錄棠众，執(zhí)行./logstash-plugin list --group input烟馅，查看是否已安裝logstash_input_jdbc插件
chenyaleideMacBook-Pro:bin lay$ ./logstash-plugin list --group input
logstash-input-beats
logstash-input-elasticsearch
logstash-input-exec
logstash-input-file
logstash-input-ganglia
logstash-input-gelf
logstash-input-generator
logstash-input-graphite
logstash-input-heartbeat
logstash-input-http
logstash-input-http_poller
logstash-input-imap
logstash-input-jdbc
logstash-input-kafka
logstash-input-pipe
logstash-input-rabbitmq
logstash-input-redis
logstash-input-s3
logstash-input-snmptrap
logstash-input-sqs
logstash-input-stdin
logstash-input-syslog
logstash-input-tcp
logstash-input-twitter
logstash-input-udp
logstash-input-unix
未安裝插件的情況下執(zhí)行./logstash-plugin install logstash-input-jdbc
3.2 安裝logstash-output-elasticsearch插件
進入logstash bin目錄说庭，執(zhí)行./logstash-plugin list --group output，查看是否已安裝logstash_output_elasticsearch插件
chenyaleideMacBook-Pro:bin lay$ ./logstash-plugin list --group output
logstash-output-cloudwatch
logstash-output-csv
logstash-output-elasticsearch
logstash-output-email
logstash-output-file
logstash-output-graphite
logstash-output-http
logstash-output-lumberjack
logstash-output-nagios
logstash-output-null
logstash-output-pagerduty
logstash-output-pipe
logstash-output-rabbitmq
logstash-output-redis
logstash-output-s3
logstash-output-sns
logstash-output-sqs
logstash-output-stdout
logstash-output-tcp
logstash-output-udp
logstash-output-webhdfs
未安裝插件的情況下執(zhí)行./logstash-plugin install logstash_output_elasticsearch
3.3 創(chuàng)建mysql測試數(shù)據(jù)
本文為方便起見郑趁，創(chuàng)建teacher與student兩張表刊驴，旨在展示如何同時同步多張表數(shù)據(jù)。
create table teacher (
id varchar(10),
first_name varchar(20),
last_name varchar(20),
age int(10),
about varchar(100),
interests varchar(100),
updatetime timestamp null default current_timestamp on update current_timestamp );
create table student (
id varchar(10),
first_name varchar(20),
last_name varchar(20),
age int(10),
about varchar(100),
interests varchar(100),
updatetime timestamp null default current_timestamp on update current_timestamp );
3.4 創(chuàng)建logstash作業(yè)文件
在/bin目錄下新建jdbc.conf文件，當(dāng)然你可以在其他目錄創(chuàng)建捆憎。
輸入以下內(nèi)容：
input {
jdbc {
# mysql jdbc connection string to our backup databse
jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/estest"
# the user we wish to excute our statement as
jdbc_user => "root"
jdbc_password => "123456"
# the path to our downloaded jdbc driver
jdbc_driver_library => "/Users/lay/.m2/repository/mysql/mysql-connector-java/5.1.37/mysql-connector-java-5.1.37.jar"
# the name of the driver class for mysql
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
jdbc_default_timezone => "Asia/Shanghai"
#可以將sql單獨提成一個文件舅柜，寫法如下
#statement_filepath => "jdbc.sql"
statement => "select * from teacher where update_time > :sql_last_value"
use_column_value => false
clean_run => true
last_run_metadata_path => "./teacher_last_run"
schedule => "* * * * *"
type => "jdbc"
#通用屬性tags
tags => ["teacher"]
}
jdbc {
# mysql jdbc connection string to our backup databse
jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/estest"
# the user we wish to excute our statement as
jdbc_user => "root"
jdbc_password => "123456"
# the path to our downloaded jdbc driver
jdbc_driver_library => "/Users/lay/.m2/repository/mysql/mysql-connector-java/5.1.37/mysql-connector-java-5.1.37.jar"
# the name of the driver class for mysql
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
jdbc_default_timezone => "Asia/Shanghai"
#statement_filepath => "jdbc.sql"
statement => "select * from student where update_time > :sql_last_value"
use_column_value => false
clean_run => true
last_run_metadata_path => "./student_last_run"
schedule => "* * * * *"
type => "jdbc"
tags => ["student"]
}
}
output {
if "teacher" in [tags]{
elasticsearch{
hosts => "127.0.0.1:9200"
index => "teacher"
document_id => "%{id}"
}
}else if "student" in [tags]{
elasticsearch{
hosts => "127.0.0.1:9200"
index => "student"
document_id => "%{id}"
}
}
stdout {
codec => json_lines
}
}
3.5 同步數(shù)據(jù)
啟動logstash服務(wù)，執(zhí)行如下命令：
logstash -f jdbc.conf
后臺啟動命令：
nohup ./logstash -f jdbc.conf > logstash.log &
數(shù)據(jù)庫插入數(shù)據(jù)：
INSERT INTO student(id,first_name,last_name,age,about,interests) VALUES('001','John','Smith', 25, 'I love to go rock climbing','[ "sports", "music" ]');
INSERT INTO student(id,first_name,last_name,age,about,interests) VALUES('002','Jane','Smith', 32, 'I like to collect rock albums','[ "music" ]');
INSERT INTO student(id,first_name,last_name,age,about,interests) VALUES('003','Douglas','Fir', 35, 'I like to build cabinets','[ "forestry" ]');
INSERT INTO teacher(id,first_name,last_name,age,about,interests) VALUES('001','John','Smith', 25, 'I love to go rock climbing','[ "sports", "music" ]');
INSERT INTO teacher(id,first_name,last_name,age,about,interests) VALUES('002','Jane','Smith', 32, 'I like to collect rock albums','[ "music" ]');
INSERT INTO teacher(id,first_name,last_name,age,about,interests) VALUES('003','Douglas','Fir', 35, 'I like to build cabinets','[ "forestry" ]');
3.6 結(jié)果驗證
查詢Elasticsearch中是否存在mysql的插入數(shù)據(jù)躲惰，至此數(shù)據(jù)同步完成致份。
