一.虚拟机下安装Elasticsearch（不在docker中安装）
1.可以选择下载或者上传Elasticsearch的安装包（是压缩文件)，放到opt目录下方，目录自己随便选择，然后进入opt目录，执行解压命令。
2.因为安全问题，在linux中不能使用root运行Elasticsearch
3。在opt下方 把Elasticsearch-7.4.0文件 授权给用户：
4.授权以后修改Elasticsearch-7.4.0文件的配置文件
进入Elasticsearch-7.4.0/config/文件
然后ls查看到elasticsearch.yml配置文件 然后执行vim命令
5.普通用户对创建文件数目有限制，内存也太小 需要设置
6.注意用户 启动Elasticsearch
7.访问地址 http://192.168.0.113:9200 可能访问不到
因为防火墙的原因，要么开发端口，要么关闭防火墙。
**************************** 结束 **********************************
二.Elasticsearch的集群（如果不选择集群 到上面步骤就可以结束了）
1.创建集群（很多个Elasticsearch服务器） 上面已经配置了一个Elasticsearch服务器，所以使用copy来拷贝更多的服务器，这里使用三个服务器。
2.修改配置文件
2.1).修改配置文件之前创建一个存放日志已经数据的文件夹,并授权给用户
2.2).修改三个的配置文件
修改配置文件：这三个服务器端口号分别为9201 9202 9203
三个服务器内部同学的端口分别为9700 9800 9900
主要是配置 http.port transport.tcp.port
服务器1的配置：
服务器二的配置：
服务器3的配置：
3.给用户授权
可以查看日志是否授权成功，一般情况都授权成功了，因为是在Elasticsearch下面的文件。直接cd /opt/elasticsearch-7.4.0-icoding1.log 然后ll查看是否授权。未授权执行下面命令。
4.修改jvm的内存，因为每台服务器默认是1G内存，这样太大了。
5.依次启动三个服务器 在对应的文件夹下面的bin目录启动
6.查看是否成功
三.安装可视化工具Kibana
1.可以选择下载或直接把压缩包拉到opt/目录下面
解压 修改配置
2.启动kibana
四.操作Kibana可视化工具
1.下图这个就是监控Elasticsearch的
2.进行创建index之类的。
3.PUT添加索引 添加type（映射）
GET查看索引
DELATE 删除索引
4.操作文档.
type的类型创建之后不能修改类型了，如果非要修改就需要重新添加一个字段，然后把数据导进去。
五.安装IK中文分析器（docker下面）
1.下载IK分词器：elasticsearch-analysis-ik-6.8.1.zip
下面是下载地址 选择对应的版本 要跟elasticsearch版本一致
https:///medcl/elasticsearch-analysis-ik/releases/download/v6.6.0/elasticsearch-analysis-ik-6.8.1.zip
2.进入容器：
docker exec -it es /bin/bash
3.在plugins目录下创建ik文件夹：
mkdir /usr/share/elasticsearch/plugins/ik
4.退出容器：exit
5.拷贝下载好的ik分词器压缩包到ik文件夹中：
docker cp /orcas/software/elasticsearch-analysis-ik-6.8.1.zip es:/usr/share/elasticsearch/plugins/ik/
6.重新进入容器 :docker exec -it es /bin/bash
进入ik目录：
cd /usr/share/elasticsearch/plugins/ik
7.解压：
unzip elasticsearch-analysis-ik-6.8.1.zip
rm -rf elasticsearch-analysis-ik-6.8.1.zip
8.exit 重启容器
docker restart es
六.Logstash数据导入（从mysql导入到ES）
1.去https:///cn/downloads/past-releases下载对应的版本。
注意注意：这样下载非常的慢
https://www.newbe.pro/Mirrors/Mirrors-Logstash/选择对应的版本 非常的快
2.icoding_course.jeson 存放与数据库的映射关系
mysql.conf：哪个数据库的表存放到哪个ES的index下面
logstash_metadata：时间戳 当前时间在这个时间之后则执行
2.wind Logstash的使用
ES的index以及type我们都创建好，以及映射。
PUT /test/
{
"mappings":{
"jd": {
"properties": {
"id": {
"type": "long"
},
"title": {
"type": "text",
"search_analyzer": "ik_max_word"
"analyzer": "ik_max_word"
},
"image":{
"type": "keyword"
},
"priceMin":{
"type": "double"
},
"priceMax":{
"type": "double"
},
"produceSrc":{
"type": "keyword"
},
"shopName": {
"type": "text",
"analyzer": "ik_smart"
"search_analyzer": "ik_smart"
},
"shopNameUri":{
"type":"keyword"
},
"commitCount":{
"type":"long"
},
"goodCount":{
"type":"long"
},
"badCount":{
"type":"long"
}
}
}
}
}
1.在Logstash的bin目录下创建一个mysql文件夹 （里面放mysql的jar以及空的last_id.txt）
2.在Logstash的config目录下面创建一个jdbc.conf
input {
stdin {}
jdbc {
type => "jdbc"
# 数据库连接地址
jdbc_connection_string => "jdbc:mysql://localhost:3306/data_shop"
# 数据库连接账号密码；
jdbc_user => "root"
jdbc_password => "2452952178"
# MySQL依赖包路径；
jdbc_driver_library => "E:/Computer/java/ES/logstash-6.8.1/bin/mysql/mysql-connector-java-5.1.46-bin.jar"
# the name of the driver class for mysql
jdbc_driver_class => "com.mysql.jdbc.Driver"
# 数据库重连尝试次数
connection_retry_attempts => "3"
# 判断数据库连接是否可用，默认false不开启
jdbc_validate_connection => "true"
# 数据库连接可用校验超时时间，默认3600S
jdbc_validation_timeout => "3600"
# 开启分页查询（默认false不开启）；
jdbc_paging_enabled => "true"
# 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
jdbc_page_size => "500"
# statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
# sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
# statement_filepath => "mysql/jdbc.sql"
#statement => "SELECT KeyId,TradeTime,OrderUserName,ModifyTime FROM `DetailTab` WHERE ModifyTime>= :sql_last_value order by ModifyTime asc"
statement => "SELECT * FROM jd"
# 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
lowercase_column_names => false
# Value can be any of: fatal,error,warn,info,debug，默认info；
sql_log_level => warn
#
# 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
record_last_run => true
# 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
use_column_value => true
# 需要记录的字段，用于增量同步，需是数据库字段
tracking_column => "id"
# Value can be any of: numeric,timestamp，Default value is "numeric"
tracking_column_type => numeric
# record_last_run上次数据存放位置；
last_run_metadata_path => "E:/Computer/java/ES/logstash-6.8.1/bin/mysql/last_id.txt"
# 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
clean_run => false
#
# 同步频率(分 时 天 月 年)，默认每分钟同步一次；
schedule => "* * * * *"
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
# convert 字段类型转换，将字段TotalMoney数据类型改为float；
mutate {
convert => {
#"TotalMoney" => "float"
}
}
}
output {
elasticsearch {
# host => "192.168.0.113"
# port => "9200"
# 配置ES集群地址
hosts => "192.168.0.113:9200"
# 索引名字，必须小写
index => "test"
# type的名字
document_type => "jd"
# 数据唯一索引（建议使用数据库KeyID）
document_id => "%{id}"
}
stdout {
codec => json_lines
}
}