需要安装

mysql数据导入ES中，用到的有logstash，ES，mysql的连接包。链接：https://pan.baidu.com/s/1xopMMUtPir12zrQfYcwBZg
提取码：me8g。这里的logstash和ES我用的版本是6.4.3的，还有mysql连接包，ik分词器6.4.3版本，数据库数据。

一. 安装ES

1.把elasticsearch-6.4.3.tar.gz安装包解压到自己熟悉的路径。

tar -zxvf elasticsearch-6.4.3.tar.gz -C /usr/local/

2.修改 elasticsearch-6.4.3/config文件夹下的elasticsearch.yml文件

cluster.name: an-es  # 集群名字，在ES集群中用的到node.name: es-node1 # ES的节点名称，ES集群中，每个几点名称不一样path.data: /usr/local/elasticsearch-6.4.3/data # data文件夹自行创建,在elasticsearch-7.4.2文件夹下path.logs: /usr/local/elasticsearch-6.4.3/logs # 日志文件存储位置network.host: 0.0.0.0 # 测试时可供提供所有访问http.port: 9200 # 端口号cluster.initial_master_nodes: ["es-node1"] # ES集群情况下节点名称

3.修改elasticsearch-6.4.3/config文件夹下jvm.xml配置文件（根据自己服务器情况配置，测试环境小的就行）

 -Xms128m-Xmx128m

4.添加用户

由于ES启动不能用root用户启动，所以要创建一个用户

 adduser esuser

同时为elasticsearch-6.4.3文件夹下的所有文件改成esuser权限

chown -R esuser:esuser elasticsearch-6.4.3/

5.修改limits.conf 和sysctl.conf配置文件

以root用户的权限修改limits.conf

 vim /etc/security/limits.conf
# 文本末尾添加这样的配置
* hard nofile 65536
* soft nofile 131072
* hard nproc 4096
* soft nproc 2048

修改limits.conf

vim /etc/sysctl.conf
#添加配置
vm.max_map_count=655360

最后刷新应用配置

sysctl -p

6.启动ES

以非root用户后台启动

/usr/local/elasticsearch-6.4.3/bin/elasticsearch -d

7.安装ES head

我直接是是在google浏览器安装的，这样的安装方式比较方便。

1.打开扩展程序

2.找到应用商店搜索

直接安装就可以了

最后连接到自己的ES就可以了

8.ik分词器安装

解压ik分词器

unzip /opt/software/elasticsearch-analysis-ik-7.4.2.zip -d /usr/local/elasticsearch-7.4.2/plugins/ik

重启ES，完成设置

二.logstash安装

1.把logstash解压到熟悉的路径

tar -zxvf logstash-6.4.3.tar.gz -C /usr/local/

2.在logstash-6.4.3文件下创建sync文件夹

mkdir sync

3.数据库数据准备

在文章开头，云盘里的foodie-shop-dev.sql文件，导入到自己的数据库即可。

4.完成logstash和数据库连接的配置文件logstash-db-sync.conf

在sync创建文件logstash-db-sync.conf，内容如下

input{
stdin{}
jdbc{
#数据库连接地址
jdbc_connection_string=>"jdbc:mysql://192.168.1.101:3306/foodie-shop-dev?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true"
##数据库连接账号密码；
jdbc_user=>"root"
jdbc_password=>"Aa_123456"
##MySQL依赖包路径；
jdbc_driver_library=>"/opt/software/mysql-connector-java-5.1.27-bin.jar"
##thenameofthedriverclassformysql
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；
##执行SQL文件的路径
statement_filepath=>"/usr/local/logstash-6.4.3/sync/foodie-items.sql"
##statement=>"SELECTKeyId,TradeTime,OrderUserName,ModifyTimeFROM`DetailTab`WHEREModifyTime>=:sql_last_valueorderbyModifyTimeasc"
##是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
lowercase_column_names=>false
##Valuecanbeanyof: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=>"updated_time"
##Valuecanbeanyof:numeric,timestamp，Defaultvalueis"numeric"
tracking_column_type=>timestamp
##record_last_run上次数据存放位置；
last_run_metadata_path=>"/usr/local/logstash-6.4.3/sync/track_time"
##是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
clean_run=>false
##
##同步频率(分时天月年)，默认每分钟同步一次；
schedule=>"* * * * *"
##索引类型
type=>"_doc"
}
}
output{
elasticsearch{
##host=>"192.168.1.1"
##port=>"9200"
##配置ES地址
hosts=>["192.168.1.101:9200"]
##索引名字，必须小写
index=>"foodie-items-ik"
##数据唯一索引（建议使用数据库KeyID）
document_id=>"%{itemId}"
# 定义模板名称
template_name=>"myik"
# 模板所在位置
template=>"/usr/local/logstash-6.4.3/sync/logstash-ik.json"
# 重写模板
template_overwrite=>true
# 默认为true，false关闭ogstash自动管理模板功能，如果自定义模板，则设置为false
manage_template=>false
}
stdout{
codec=>json_lines
}
}

在这里需要数据库连接包，准备好放入到sync文件夹下。这里的主要对数据库中items表的数据进行导入，所以需要一个sql语句，在sync文件夹下创建foodie-items.sql。这里根据自己的需要写SQL

5.创建外部文件SQL

foodie-items.sql

SELECTi.id AS itemId,i.item_name AS itemName,i.sell_counts AS sellCounts,ii.url AS imgUrl,tempSpec.priceDiscount price,i.updated_time as updated_time
FROMitems i
LEFT JOIN items_img ii ON i.id = ii.item_id
LEFT JOIN (SELECTitem_id,MIN(price_discount) priceDiscountFROMitems_specGROUP BYitem_id
) tempSpec ON i.id = tempSpec.item_id
WHEREii.is_main = 1and i.updated_time >= :sql_last_value

这里的主键是itemId，要对应到输出的ES，document_id=>"%{itemId}"，是es的id。i.updated_time >= :sql_last_value这个是在更新数据库的时候同时ES中的数据也更新。ES的更新是根据items表中的updated_time更新而更新的。只有当updated_time大于sync文件夹下的track_time时，才进行ES中的数据更新。track_time文件内如如下

--- 2020-04-29 14:45:39.000000000 +08:00

所以，只有数据库表中的updated_time大于track_time里的时间，才进行更新。

6.加入ik分词器

在logstash-db-sync.conf文件中，需要一个模板，在sync下创建

logstash-ik.json

{"order": 0,"version": 1,"index_patterns": ["*"],"settings": {"index": {"refresh_interval": "5s"}},"mappings": {"_default_": {"dynamic_templates": [{"message_field": {"path_match": "message","match_mapping_type": "string","mapping": {"type": "text","norms": false}}},{"string_fields": {"match": "*","match_mapping_type": "string","mapping": {"type": "text","norms": false,"analyzer": "ik_max_word","fields": {"keyword": {"type": "keyword","ignore_above": 256}}}}}],"properties": {"@timestamp": {"type": "date"},"@version": {"type": "keyword"},"geoip": {"dynamic": true,"properties": {"ip": {"type": "ip"},"location": {"type": "geo_point"},"latitude": {"type": "half_float"},"longitude": {"type": "half_float"}}}}}},"aliases": {}}

这个文件是从ES中取来的，然后做一下修改

修改

1."index-patterns":["*"]，对所有的索引进行配置；

2."version":1，版本从1开始

3.string_fields下的mapping中加入"analyzer": "ik_max_word"，配置分词器

4.去掉最外层的{"logstash": }。

看一下sync下的都有什么文件

7.启动logstash

进入目录

./logstash -f /usr/local/logstash-6.4.3/sync/logstash-db-sync.conf

出现这样的情况，数据已经同步

8.遇到的问题

再做ik分词器模板时，以上面的方式，一直ik分词器不能生效。我把ik分词器的文件放到postman里，手动的生成了一下。生成一个myik的模板，在制定ik分词器的时候，名字叫template_name: myik。有点头疼

MySQL数据库数据导入ES中相关推荐

将数据库的数据导入es中
数据量小的方法一: 首先必须创建client public class ClientConfig {/*** 批量导入到es中必须的* @return RestHighLevelClient*/@Be ...
【C#】MySQL数据库数据导入（批量Excel插入）
系列文章 [C#]代码模板生成工具本文链接:https://blog.csdn.net/youcheng_ge/article/details/126890673 [C#]MySQL数据库导入工具( ...
往远程mysql传数据库_三、Navicat将远程MySql数据库数据导入本地
1.安装本地的MySql.记住用户名和密码,这里以root,root为例. 2.打开Navicat,新建连接(连接),输入连接名,用户名,密码.确定,连接测试.这里连接名为luzhanshi.这样本地 ...
cdh mysql sqoop 驱动_[bigdata-003]在cdh 5.7下用sqoop1将mysql数据库数据导入到hdfs的方式...
1. 假设,myql安装在bigdata2上.我们要在bigdata3上执行sqoop1. 2. 首先,要在mysql上创建一个'b3'@'%'的账户,这个账户限定只能从外部ip地址访问mysql. ...
mysql导入es_mysql数据导入es
1,py-mysql-elasticsearch-sync<github 工具> 简介: python版本的mysql导入es的插件通过配置config.yaml文件,该插件可以将mys ...
【Mysql】Mysql数据库数据导出\导入
如何在 MySQL / MariaDB 中导入导出数据,导入导出数据库文件.Excel.CSV - 卡拉云本文介绍Mysql数据库数据导出及导入方法.以及重要补充内容目录 1. MySQL 数据库 ...
效率最高的Excel数据导入---（c#调用SSIS Package将数据库数据导入到Excel文件中【附源代码下载】）...
本文目录: (一)背景 (二)数据库数据导入到Excel的方法比较 (三)SSIS的简介 (四)数据库中存储过程示例(SSIS应用需要) (五)Excel模板的制作(这步这么简单,稍微介 ...
用sqoop将mysql的数据导入到hive表中，原理分析
Sqoop 将 Mysql 的数据导入到 Hive 中准备Mysql 数据如图所示,准备一张表,数据随便造一些,当然我这里的数据很简单. 编写命令编写参数文件个人习惯问题,我喜欢把参数写到文件 ...
php如何查询数据库,如何在php中查询mysql数据库数据
如何在php中查询mysql数据库数据发布时间:2020-07-21 09:23:55 来源:亿速云阅读:81 作者:Leah 本篇文章给大家分享的是有关如何在php中查询mysql数据库数据,小 ...

MySQL数据库数据导入ES中