Detailed explanation of how to synchronize data from MySQL to Elasticsearch

cd /logstash-6.4.2/bin
./logstash-plugin install logstash-input-jdbc

select * from t_employee

cd logstash-6.4.2
# Check whether the configuration file syntax is correct bin/logstash -f config/jdbc.conf --config.test_and_exit
# Start bin/logstash -f config/jdbc.conf --config.reload.automatic

PUT octopus
GET octopus/_search

ROW: records each row of data that has been modified, but the log volume is too large. STATEMENT: records each SQL statement that modifies the data, which reduces the log volume, but the SQL statement is prone to master-slave inconsistency when using functions or triggers. MIXED: combines the advantages of ROW and STATEMENT, and chooses to use ROW or STATEMENT to record logs according to the specific SQL statement that performs data operations.

SET TIMESTAMP=1527917394/*!*/;
    BEGIN
    /*!*/;
    # at 3751
    #180602 13:29:54 server id 1 end_log_pos 3819 CRC32 0x8dabdf01 Table_map: `webservice`.`building` mapped to number 74
    # at 3819
    #180602 13:29:54 server id 1 end_log_pos 3949 CRC32 0x59a8ed85 Update_rows: table id 74 flags: STMT_END_F
    
    BINLOG '
    UisSWxMBAAAARAAAAOsOAAAAAEoAAAAAAAEACndlYnNlcnZpY2UACGJ1aWxkaW5nAAYIDwEPEREG
    wACAAQAAAAHfq40=
    UisSWx8BAAAAggAAAG0PAAAAAEoAAAAAAAEAAgAG///A1gcAAAAAAAALYnVpbGRpbmctMTAADwB3
    UkRNbjNLYlV5d1k3ajVbD64WWw+uFsDWBwAAAAAAAAtidWlsZGluZy0xMAEPAHdSRE1uM0tiVXl3
    WTdqNVsPrhZbD64Whe2oWQ==
    '/*!*/;
    ### UPDATE `webservice`.`building`
    ###WHERE
    ### @1=2006 /* LONGINT meta=0 nullable=0 is_null=0 */
    ### @2='building-10' /* VARSTRING(192) meta=192 nullable=0 is_null=0 */
    ### @3=0 /* TINYINT meta=0 nullable=0 is_null=0 */
    ### @4='wRDMn3KbUywY7j5' /* VARSTRING(384) meta=384 nullable=0 is_null=0 */
    ### @5=1527754262 /* TIMESTAMP(0) meta=0 nullable=0 is_null=0 */
    ### @6=1527754262 /* TIMESTAMP(0) meta=0 nullable=0 is_null=0 */
    ### SET
    ### @1=2006 /* LONGINT meta=0 nullable=0 is_null=0 */
    ### @2='building-10' /* VARSTRING(192) meta=192 nullable=0 is_null=0 */
    ### @3=1 /* TINYINT meta=0 nullable=0 is_null=0 */
    ### @4='wRDMn3KbUywY7j5' /* VARSTRING(384) meta=384 nullable=0 is_null=0 */
    ### @5=1527754262 /* TIMESTAMP(0) meta=0 nullable=0 is_null=0 */
    ### @6=1527754262 /* TIMESTAMP(0) meta=0 nullable=0 is_null=0 */
    # at 3949
    #180602 13:29:54 server id 1 end_log_pos 3980 CRC32 0x58226b8f Xid = 182
    COMMIT /*!*/;

SET TIMESTAMP=1527919329/*!*/;
    update building set Status=1 where Id=2000
    /*!*/;
    # at 688
    #180602 14:02:09 server id 1 end_log_pos 719 CRC32 0x4c550a7d Xid = 200
    COMMIT /*!*/;

# View binlog modemysql> show variables like "%binlog_format%";
 
# Modify binlog modemysql> set global binlog_format='ROW';
 
# Check whether binlog is enabled mysql> show variables like 'log_bin';
 
# Enable binlog
Modify the my.cnf file log-bin = mysql-bin

# Install go
sudo apt-get install go
 
# Install godep
go get github.com/tools/godep
 
# Get the go-mysql-elasticsearch plugin go get github.com/siddontang/go-mysql-elasticsearch
 
# Install the go-mysql-elasticsearch plugin cd go/src/github.com/siddontang/go-mysql-elasticsearch
make

# MySQL address, user and password
# user must have replication privilege in MySQL.
my_addr = "127.0.0.1:3306" # Basic mysql settings that need to be synchronized my_user = "root"
my_pass = "root"
 
# Elasticsearch address
es_addr = "127.0.0.1:9200" # Local elasticsearch configuration # Path to store data, like master.info, and dump MySQL data 
data_dir = "./var" # data storage url
# The following configuration is saved as default# Inner Http status address
stat_addr = "127.0.0.1:12800"
 
# pseudo server id like a slave 
server_id = 1001
 
#mysql or mariadb
flavor = "mysql"
#mysqldump execution path
mysqldump = "mysqldump"
 
#MySQL data source
[[source]]
schema = "test" //The corresponding database name when elasticsearch is synchronized with mysql # Only below tables will be synced into Elasticsearch.
# To synchronize several tables in the test database. For some projects that use the table partitioning mechanism, we can use wildcards to match, for example, t_[0-9]{4} can match table t_0000 to t_9999.
tables = ["t", "t_[0-9]{4}", "tfield", "tfilter"]  
 
# Below is for special rule mapping
# For a table, we need to specify the type of the ES index to which its data should be synchronized. If not specified, we will use schema # name as the ES index and type by default
[[rule]]
schema = "test" //Database nametable = "t" //Table nameindex = "test" //Corresponding index nametype = "t" //Corresponding type name# Synchronize all tables that meet the format of t_[0-9]{4} to the ES with index test and type t. Of course, these tables need to ensure that the # schema is consistent [[rule]]
schema = "test"
table = "t_[0-9]{4}"
index = "test"
type = "t"
 
# For table tfilter, we will only synchronize the id and name columns, and the others will not be synchronized filter = ["id", "name"]
# The column id of table tfield is mapped to es_id, and tags are mapped to es_tags
# list This field explicitly tells you that the corresponding column data needs to be converted into ES array type. This is now commonly used for MySQL's varchar # and other types. We may store data like "a, b, c", and then hope to convert it into a list format like [a, b, c] when synchronizing it to ES.
 
[rule.field]
# Map column `id` to ES field `es_id`
id="es_id"
# Map column `tags` to ES field `es_tags` with array type 
tags="es_tags,list"
# Map column `keywords` to ES with array type
keywords=",list"

cd go/src/github.com/siddontang/go-mysql-elasticsearch
bin/go-mysql-elasticsearch -config=./etc/river.toml

[root@autofelix /]# vi /etc/profile
export JDBC_IMPORTER_HOME=/elasticsearch-jdbc-2.3.2.0

[root@autofelix /]# source /etc/profile

[root@autofelix /]# ll /odbc_es/
drwxr-xr-x 2 root root 4096 Jun 16 03:11 logs
-rwxrwxrwx 1 root root 542 Jun 16 04:03 mysql_import_es.sh

[root@autofelix odbc_es]# cat mysql_import_es.sh
'#!/bin/sh
bin=$JDBC_IMPORTER_HOME/bin
lib=$JDBC_IMPORTER_HOME/lib
echo '{
"type" : "jdbc",
"jdbc": {
"elasticsearch.autodiscover":true,
"elasticsearch.cluster":"my-application", #Cluster name, see: /usr/local/elasticsearch/config/elasticsearch.yml
"url":"jdbc:mysql://10.8.5.101:3306/test", #mysql database address "user":"root", #mysql username "password":"123456", #mysql password "sql":"select * from cc",
"elasticsearch" : {
  "host" : "10.8.5.101",
  "port" : 9300
},
"index" : "myindex", #new index
"type" : "mytype" #new type
}
}'| java \
  -cp "${lib}/*" \
  -Dlog4j.configurationFile=${bin}/log4j2.xml \
  org.xbib.tools.Runner \
  org.xbib.tools.JDBCImporter

[root@autofelix odbc_es]# chmod a+x mysql_import_es.sh

[root@autofelix odbc_es]# ./mysql_import_es.sh