Elasticsearch做搜索服务
阅读本文需要有以下基础
如果不知道ELK的安装 请看：ELK日志收集
如果不清楚logstash的配置请参见：logstash应用解耦合
简介
所谓使用Elasticsearch做搜索服务，就是数据量大的时候，数据库无法承载压力，可以将数据放到Elasticsearch做索引，当需要查询的时候直接从Elasticsearch取数据。
步骤
1 Logstash从mysql取出数据
2 Logstash发送数据到Elasticsearch
3 从Elasticsearch查询数据
这里主要的配置在Logstash
配置Logstash
全量数据同步
input {
jdbc {
# mysql连接jar包位置
jdbc_driver_library => "/path/mysql-connector-java-6.0.6.jar"
# 连接类 5.*版本为com.mysql.jdbc.Driver 6.*版本为com.mysql.cj.jdbc.Driver
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
# 数据库链接
jdbc_connection_string => "jdbc:mysql://localhost:3306/test?characterEncoding=UTF-8&useSSL=false"
jdbc_user => "root"
jdbc_password => "password"
statement => "SELECT * FROM company_info"
jdbc_paging_enabled => "true"
jdbc_page_size => "5000"
schedule => "*/10 * * * *"
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
}
output {
elasticsearch {
hosts => ["http://localhost:9200"]
# COMPANY_ID为上表的主键
document_id => "%{COMPANY_ID}"
index => "company_info "
}
stdout {
codec => rubydebug
}
}
增量数据同步
# 配置从mysql输入数据
input {
jdbc {
# mysql连接jar包位置
jdbc_driver_library => "/path/mysql-connector-java-6.0.6.jar"
# 连接类 5.*版本为com.mysql.jdbc.Driver 6.*版本为com.mysql.cj.jdbc.Driver
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
# 数据库链接
jdbc_connection_string => "jdbc:mysql://localhost:3306/test?characterEncoding=UTF-8&useSSL=false"
# 数据库用户
jdbc_user => "root"
# 数据库密码
jdbc_password => "123456"
# 是否启用分页
jdbc_paging_enabled => "true"
# 每一批传输的数量
jdbc_page_size => "1000"
# 多长时间执行一次 cron表达式 以下为每分钟执行一次
schedule => "* * * * *"
# 时区
jdbc_default_timezone => "Asia/Shanghai"
# 是否保存状态
record_last_run => "true"
# true 使用定义的 tracking_column值作为:sql_last_value
# false :sql_last_value反映上次执行查询的时间
use_column_value => "true"
# 配合use_column_value使用
# 这样配置后，下面的statement选项可以使用了
tracking_column => "UPDATE_TIME"
# 记录:sql_last_value的文件
last_run_metadata_path => "/path/last_run_time.log"
# 是否将数据库中的列名自动转换为小写
lowercase_column_names => "false"
# tracking_column的数据类型，只能是numberic和timestamp
tracking_column_type => "timestamp"
# 是否应清理先前的运行状态，不清楚什么用
clean_run => "false"
# 从数据库中抓数据的SQL脚本
statement => "SELECT * FROM company_info where UPDATE_TIME > :sql_last_value"
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
}
# 配置数据输出位置
output {
elasticsearch {
hosts => ["http://localhost:9200"]
# COMPANY_ID为上表的主键
document_id => "%{COMPANY_ID}"
index => "company_info "
}
stdout {
codec => rubydebug
}
}
从Elasticsearch查询数据
curl 'localhost:9200/company_info/_search?q=*&pretty'
// 返回数据
{
"took" : 1,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"skipped" : 0,
"failed" : 0
},
"hits" : {
"total" : {
"value" : 56,
"relation" : "eq"
},
"max_score" : 1.0,
"hits" : [
{
"_index" : "share_sys_company_info",
"_type" : "_doc",
"_id" : "788e3fd383344e7eb4612412661b1676",
"_score" : 1.0,
#~~~~~~~~~~~ 这就是我数据库中的数据 ~~~~~~~~~~~~~~~
"_source" : {
"BUSSINESS_TYPE" : "4bdbdb5beff946a3ab161fbf18ef6145",
"COMPANY_ROLE" : "",
"@version" : "1",
"SHORT_NAME" : "无锡换热器厂",
"COMPETENCY_TYPE" : "2",
"PARENT_COMPANY" : "-1",
"NETWORK_STATUS" : null,
"GOVERN_TYPE" : null,
"CREATE_TIME" : "2019-03-13T09:40:42.000Z",
"UPDATE_USER" : "99408e75541f425886dc50c3186292eb",
"COMPANY_ID" : "788e3fd383344e7eb4612412661b1676",
"UPDATE_TIME" : "2019-04-16T11:11:15.000Z",
"LOGO" : null,
"NAME" : "无锡市换热器厂",
"DEL_FLAG" : "N",
"BIN_CODE" : "58821",
"URL" : null,
"NETWORK_EXPIRED" : null,
"@timestamp" : "2019-07-30T08:18:02.096Z",
"MARK" : "29",
"NETWORK_DATE" : null,
"NETWORK_SIGN" : null,
"PIDS" : "-1",
"GROUP_TYPE" : "b77c305c79294a26a230b16e8ced3235",
"CREATE_USER" : "99408e75541f425886dc50c3186292eb",
"CORP_TYPE" : "150df743bc134e47be7b7d97ad8e3adb",
"LOGO_PRO" : "c340f509b2b54b82accbf4dba63c2cca",
"ORDERED" : 29
}
}
}
]
}
}
参考
https://www.elastic.co/guide/cn/elasticsearch/guide/current/query-dsl-intro.html
https://juejin.im/post/5b7fe4a46fb9a019d92469a9#heading-11
https://www.cnblogs.com/gqhwk/p/9440787.html