本文节选自《Netkiller Database 手札》
MySQL 导入 Elasticsearch 的方法有很多,通常是使用ETL工具,但我觉得太麻烦。于是想到 logstash 。
安装 JDBC 驱动 和 Logstash
curl -s https://raw.githubusercontent.com/oscm/shell/master/database/mysql/5.7/mysql-connector-java.sh | bash
curl -s https://raw.githubusercontent.com/oscm/shell/master/log/kibana/logstash-5.x.sh | bash
mysql 驱动文件位置在 /usr/share/java/mysql-connector-java.jar
创建配置文件 /etc/logstash/conf.d/jdbc-mysql.conf
mysql> desc article;
+-------------+--------------+------+-----+---------+-------+
| Field | Type | Null | Key | Default | Extra |
+-------------+--------------+------+-----+---------+-------+
| id | int(11) | NO | | 0 | |
| title | mediumtext | NO | | NULL | |
| description | mediumtext | YES | | NULL | |
| author | varchar(100) | YES | | NULL | |
| source | varchar(100) | YES | | NULL | |
| ctime | datetime | NO | | NULL | |
| content | longtext | YES | | NULL | |
+-------------+--------------+------+-----+---------+-------+
7 rows in set (0.00 sec)
input {
jdbc {
jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/cms"
jdbc_user => "cms"
jdbc_password => "password"
schedule => "* * * * *"
statement => "select * from article"
}
}
output {
elasticsearch {
hosts => "localhost:9200"
index => "information"
document_type => "article"
document_id => "%{id}"
}
}
root@netkiller /var/log/logstash % systemctl restart logstash % curl -XGET 'http://localhost:9200/_all/_search?pretty' 适合数据没有改变的归档数据或者只能增加没有修改的数据 input { 多张数据表导入到 Elasticsearch # multiple inputs on logstash jdbc 需要在每一个jdbc配置项中加入 type 配置,然后 elasticsearch 配置项中加入 document_type => "%{type}" input { tracking_column_type => "numeric" 可以声明 id 字段的数据类型, 如果不指定将会默认为日期 [2017-07-31T11:08:00,193][INFO ][logstash.inputs.jdbc ] (0.020000s) select * from article where id > '2017-07-31 02:47:00' 如果复制不对称可以加入 clean_run => true 配置项,清楚数据 input { 如果复制不对称可以加入 clean_run => true 配置项,清楚数据 statement_filepath 指定 SQL 文件,有时SQL太复杂写入 statement 配置项维护部方便,可以将 SQL 写入一个文本文件,然后使用 statement_filepath 配置项引用该文件。 input { 将需要复制的条件参数写入 parameters 配置项 input { jdbc_fetch_size => 1000 #jdbc获取数据的数量大小 input { We accept PayPal through: https://www.paypal.me/netkiller Wechat (微信) / Alipay (支付宝) 打赏: http://www.netkiller.cn/home/donations.html 作者相关文章: Spring data 数据库建表(一对一,一对多,多对多) Apache Sqoop 将mysql导入到Hadoop HDFS Spring boot with Apache Hive Apache Hive 快速入门 CentOS 7.3 + Server JRE 1.8 + Hadoop-2.8.0 Apache Hbase 快速入门 Spring cloud 之 Feign Client Spring Cloud Netflix 重新整理AUTO_INCREMENT字段 Spring Cloud Config Spring boot with Schedule (启用/禁用) Spring boot with HTTPS SSL Spring boot with Git version 《Netkiller Virtualization 手札》Docker 卷管理 PHP高级编程之守护进程 Spring boot with Docker Spring boot with Service Spring boot with PostgreSQL Struts2 S2-046, S2-045 Firewall(漏洞防火墙) 数据库与图片完美解决方案 数据库进程间通信解决方案 数据库进程间通信解决方案之MQ Linux 系统安全与优化配置 Tomcat 安全配置与性能优化 Linux 系统与数据库安全 转载请注明出处与作者声明,扫描二维码关注作者公众好,不定期更新文章
root@netkiller /var/log/logstash % systemctl status logstash
● logstash.service - logstash
Loaded: loaded (/etc/systemd/system/logstash.service; enabled; vendor preset: disabled)
Active: active (running) since Mon 2017-07-31 09:35:00 CST; 11s ago
Main PID: 10434 (java)
CGroup: /system.slice/logstash.service
└─10434 /usr/bin/java -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -XX:+UseCMSInitiatingOccupancyOnly -XX:+DisableExplicitGC -Djava.awt.headless=true -Dfi...
Jul 31 09:35:00 VM_3_2_centos systemd[1]: Started logstash.
Jul 31 09:35:00 VM_3_2_centos systemd[1]: Starting logstash...
root@netkiller /var/log/logstash % cat logstash-plain.log
[2017-07-31T09:35:28,169][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
[2017-07-31T09:35:28,172][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://localhost:9200/, :path=>"/"}
[2017-07-31T09:35:28,298][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>#
[2017-07-31T09:35:28,299][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2017-07-31T09:35:28,337][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>50001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"_default_"=>{"_all"=>{"enabled"=>true, "norms"=>false}, "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, "fields"=>{"keyword"=>{"type"=>"keyword", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date", "include_in_all"=>false}, "@version"=>{"type"=>"keyword", "include_in_all"=>false}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}}
[2017-07-31T09:35:28,344][INFO ][logstash.outputs.elasticsearch] Installing elasticsearch template to _template/logstash
[2017-07-31T09:35:28,465][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>[#
[2017-07-31T09:35:28,483][INFO ][logstash.pipeline ] Starting pipeline {"id"=>"main", "pipeline.workers"=>8, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>1000}
[2017-07-31T09:35:29,562][INFO ][logstash.pipeline ] Pipeline main started
[2017-07-31T09:35:29,700][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600}
[2017-07-31T09:36:01,019][INFO ][logstash.inputs.jdbc ] (0.006000s) select * from article
23.8.4. 验证
23.8.5. 配置模板
23.8.5.1. 全量导入
jdbc {
jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/cms"
jdbc_user => "cms"
jdbc_password => "password"
schedule => "* * * * *"
statement => "select * from article"
}
}
output {
elasticsearch {
hosts => "localhost:9200"
index => "information"
document_type => "article"
document_id => "%{id}"
}
}
23.8.5.2. 多表导入
input {
jdbc {
jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/cms"
jdbc_user => "cms"
jdbc_password => "password"
schedule => "* * * * *"
statement => "select * from article"
type => "article"
}
jdbc {
jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/cms"
jdbc_user => "cms"
jdbc_password => "password"
schedule => "* * * * *"
statement => "select * from comment"
type => "comment"
}
}
output {
elasticsearch {
hosts => "localhost:9200"
index => "information"
document_type => "%{type}"
document_id => "%{id}"
}
}
23.8.5.3. 通过 ID 主键字段增量复制数据
jdbc {
statement => "SELECT id, mycolumn1, mycolumn2 FROM my_table WHERE id > :sql_last_value"
use_column_value => true
tracking_column => "id"
tracking_column_type => "numeric"
# ... other configuration bits
}
}
23.8.5.4. 通过日期字段增量复制数据
jdbc {
statement => "SELECT * FROM my_table WHERE create_date > :sql_last_value"
use_column_value => true
tracking_column => "create_date"
# ... other configuration bits
}
}
23.8.5.5. 指定SQL文件
jdbc {
jdbc_driver_library => "/path/to/driver.jar"
jdbc_driver_class => "org.postgresql.Driver"
jdbc_url => "jdbc://postgresql"
jdbc_user => "neo"
jdbc_password => "password"
statement_filepath => "query.sql"
}
}
23.8.5.6. 参数传递
jdbc {
jdbc_driver_library => "mysql-connector-java-5.1.36-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"
jdbc_user => "mysql"
parameters => { "favorite_artist" => "Beethoven" }
schedule => "* * * * *"
statement => "SELECT * from songs where artist = :favorite_artist"
}
}
23.8.5.7. 控制返回JDBC数据量
jdbc_page_size => 1000 #jdbc一页的大小,
jdbc_paging_enabled => true #和jdbc_page_size组合,将statement的查询分解成多个查询,相当于: SELECT * FROM table LIMIT 1000 OFFSET 4000
23.8.6. example
jdbc {
jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/cms"
jdbc_user => "cms"
jdbc_password => "password"
schedule => "* * * * *" #定时cron的表达式,这里是每分钟执行一次
statement => "select id, title, description, author, source, ctime, content from article where id > :sql_last_value"
use_column_value => true
tracking_column => "id"
tracking_column_type => "numeric"
record_last_run => true
last_run_metadata_path => "/var/tmp/article.last"
}
}
output {
elasticsearch {
hosts => "localhost:9200"
index => "information"
document_type => "article"
document_id => "%{id}"
action => "update" # 操作执行的动作,可选值有["index", "delete", "create", "update"]
doc_as_upsert => true #支持update模式
}
}Donations (打赏)