共计 3783 个字符，预计需要花费 10 分钟才能阅读完成。
1、应用背景
咱们习惯性，将数据存储到 mysql，那么怎么应用 ES 实现全文检索呢。Elasticsearch 为咱们提供了一个插件 logstash 来同步咱们的数据库。
2、装置 logsash 插件
注意事项：要装置跟 es 版本一样的版本。
logstash 下载地址：https://www.elastic.co/downloads/logstash
在 windows 上面解压即可应用。我这里用的都是 7.2.0 版本。不须要再像 2.x 版本须要集成 logstash-jdbc-input，能力实现数据同步。
3、配置 logsash 实现同步
1）在根目录下新建文件夹 mysqletc，用来搁置同步所须要的文件和 mysql 驱动
文件夹内容：
2）将要同步的数据库对应的 mysql 驱动放到该文件夹
3）在创立的新文件夹中（mysqletc 文件夹）中创立 sql 文件
sql 文件里就是你须要同步到 es 中的数据的查问语句，如果是全量同步，只须要要 select * from [table]即可。
如果同步多个表格能够创立多个 sql 文件。
比方我的：
soul.sql
SELECT * FROM soulmto_user.sql
SELECT * FROM mto_user留神：结尾不要加分号，我加分号同步报错了
3）logstash 链接数据库和 Elasticsearch 的 conf 文件
名字轻易取，我这里习惯叫 mysql.conf，内容如下：
input {stdin {}
jdbc {
# mysql 数据库链接,shop 为数据库名
jdbc_connection_string => "jdbc:mysql://114.67.169.20:3306/myblog"
# 用户名和明码
jdbc_user => "root"
jdbc_password => "Wylc!@#$5"
# 驱动
jdbc_driver_library => "C:/softs/elk/logstash-7.2.0/mysqletc/mysql-connector-java-8.0.13.jar"
# 驱动类名
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
# 执行的 sql 文件门路 + 名称
statement_filepath => "C:/softs/elk/logstash-7.2.0/mysqletc/soul.sql"
# 设置监听距离 各字段含意（由左至右）分、时、天、月、年，全副为 * 默认含意为每分钟都更新
schedule => "* * * * *"
# 索引类型
type => "soul"
}
jdbc {
# mysql 数据库链接,shop 为数据库名
jdbc_connection_string => "jdbc:mysql://114.67.169.20:3306/myblog"
# 用户名和明码
jdbc_user => "root"
jdbc_password => "Wylc!@#$5"
# 驱动
jdbc_driver_library => "C:/softs/elk/logstash-7.2.0/mysqletc/mysql-connector-java-8.0.13.jar"
# 驱动类名
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
# 执行的 sql 文件门路 + 名称
statement_filepath => "C:/softs/elk/logstash-7.2.0/mysqletc/mto_user.sql"
# 设置监听距离 各字段含意（由左至右）分、时、天、月、年，全副为 * 默认含意为每分钟都更新
schedule => "* * * * *"
# 索引类型
type => "mto_user"
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
}
output {if [type]=="soul"{
elasticsearch {hosts => ["localhost:9200"]
index => "soul"
document_id => "%{id}"
}
}
if [type]=="mto_user"{
elasticsearch {hosts => ["localhost:9200"]
index => "mto_user"
document_id => "%{id}"
}
}
stdout {codec => json_lines}
}留神：创立的所有文本文件肯定要是 utf- 8 无 BOM 格局编码
实现了下面的步骤，logstash 的配置也就实现了。
4、启动 logstash 开始同步数据库
第一步：运行 elasticsearch.bat 文件，关上 elasticsearch，运行胜利能够看到：
第二步：到 bin 目录下创立一个批处理文件 run_default.bat，内容为：
logstash -f ../mysqletc/mysql.conf第三步：双击 run_default.bat，其中 logstash -f 示意运行指令，../mysqlec/mysql.conf 示意咱们配置的 mysql.conf 文件门路，胜利启动后，能够在终端中看见运行的 sql 和同步的数据
5、测试是否同步胜利
同步实现后，咱们能够去 head 插件里，查看 es 中是否存在咱们方才创立的索引库：
在根本查问中，能够看见咱们同步的数据
其中：timestamp 和 version 是 elastisearch 本人增加的字段。
相干文章：怎么装置应用 elasticsearch-head 插件
input {stdin {}
jdbc {
# mysql 数据库链接,shop 为数据库名
jdbc_connection_string => "jdbc:mysql://114.67.169.20:3306/myblog"
# 用户名和明码
jdbc_user => "root"
jdbc_password => "xxxxxx"
# 驱动
jdbc_driver_library => "C:/softs/elk/logstash-7.2.0/mysqletc/mysql-connector-java-8.0.13.jar"
# 驱动类名
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
# 执行的 sql 文件门路 + 名称
statement_filepath => "C:/softs/elk/logstash-7.2.0/mysqletc/soul.sql"
# 设置监听距离 各字段含意（由左至右）分、时、天、月、年，全副为 * 默认含意为每分钟都更新
schedule => "* * * * *"
# 索引类型
type => "soul"
}
jdbc {
# mysql 数据库链接,shop 为数据库名
jdbc_connection_string => "jdbc:mysql://114.67.169.20:3306/myblog"
# 用户名和明码
jdbc_user => "root"
jdbc_password => "xxxxxx"
# 驱动
jdbc_driver_library => "C:/softs/elk/logstash-7.2.0/mysqletc/mysql-connector-java-8.0.13.jar"
# 驱动类名
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
# 执行的 sql 文件门路 + 名称
statement_filepath => "C:/softs/elk/logstash-7.2.0/mysqletc/mto_user.sql"
# 设置监听距离 各字段含意（由左至右）分、时、天、月、年，全副为 * 默认含意为每分钟都更新
schedule => "* * * * *"
# 索引类型
type => "mto_user"
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
}
output {if [type]=="soul"{
elasticsearch {hosts => ["localhost:9200"]
index => "soul"
document_id => "%{id}"
}
}
if [type]=="mto_user"{
elasticsearch {hosts => ["localhost:9200"]
index => "mto_user"
document_id => "%{id}"
}
}
stdout {codec => json_lines}
}相干文章：