配置logstash拉取myql8.0.18資料庫中的資料到elastcisearch中
阿新 • • 發佈：2020-08-24
配置logstash拉取myql8.0.18資料庫中的資料到elastcisearch中
環境:
系統：centos7.8_x86_64
# java --version
openjdk 13.0.1 2019-10-15
1.下載並配置logstash
# 修改 logstash.yml
path.config: /usr/local/elk/logstash-7.4.2/config/logstash.conf
# cat /usr/local/elk/logstash-7.4.2/config/logstash.conf
input { stdin { } jdbc { jdbc_connection_string=> "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false" jdbc_user => "mysqluser" jdbc_password => "mysqlpass" jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" statement=> "SELECT * FROM font_element WHERE updated_at > :sql_last_value" jdbc_default_timezone =>"Asia/Shanghai" record_last_run => "true" use_column_value => "true" tracking_column => "updated_at" tracking_column_type => "timestamp" last_run_metadata_path=> "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_font_element" clean_run => "false" schedule => "*/3 * * * *" jdbc_paging_enabled => "true" jdbc_page_size => "50000" type => "font-element" } jdbc { jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false" jdbc_user => "mysqluser" jdbc_password => "mysqlpass" jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_default_timezone =>"Asia/Shanghai" statement => "SELECT i.id, i.guid, i.thumbnail, i.property, i.updated_at, i.created_at, kg.keywords, ec.category_name FROM image_element i left join ( select element_id, GROUP_CONCAT(keyword separator ' ') as keywords from image_element_keywords as ie INNER JOIN keywords as k on ie.keywords_id = k.id GROUP BY element_id ) as kg on i.id = kg.element_id left JOIN element_category as ec on i.category_id = ec.id WHERE i.updated_at > :sql_last_value" record_last_run => "true" use_column_value => "true" tracking_column => "updated_at" tracking_column_type => "timestamp" last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_image_element" clean_run => "false" schedule => "*/3 * * * *" jdbc_paging_enabled => "true" jdbc_page_size => "50000" type => "image-element" } jdbc { jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false" jdbc_user => "mysqluser" jdbc_password => "mysqlpass" jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_default_timezone =>"Asia/Shanghai" statement => "SELECT i.id, i.guid, i.thumbnail, i.property, i.updated_at, i.created_at, kg.keywords, '' as category_name FROM background_element as i left join ( select element_id, GROUP_CONCAT(keyword separator ' ') as keywords from background_element_keywords as ie INNER JOIN keywords as k on ie.keywords_id = k.id GROUP BY element_id ) as kg on i.id = kg.element_id WHERE i.updated_at > :sql_last_value" record_last_run => "true" use_column_value => "true" tracking_column => "updated_at" tracking_column_type => "timestamp" last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_background_element" clean_run => "false" schedule => "*/3 * * * *" jdbc_paging_enabled => "true" jdbc_page_size => "50000" type => "background-element" } jdbc { jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false" jdbc_user => "mysqluser" jdbc_password => "mysqlpass" jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_default_timezone =>"Asia/Shanghai" statement => "SELECT i.id, i.guid, i.thumbnail, i.property, i.updated_at, i.created_at, kg.keywords, ec.category_name FROM shape_element as i left join ( select element_id, GROUP_CONCAT(keyword separator ' ') as keywords from shape_element_keywords as ie INNER JOIN keywords as k on ie.keywords_id = k.id GROUP BY element_id ) as kg on i.id = kg.element_id left JOIN element_category as ec on i.category_id = ec.id WHERE i.updated_at > :sql_last_value" record_last_run => "true" use_column_value => "true" tracking_column => "updated_at" tracking_column_type => "timestamp" last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_shape_element" clean_run => "false" schedule => "*/3 * * * *" jdbc_paging_enabled => "true" jdbc_page_size => "50000" type => "shape-element" } jdbc { jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false" jdbc_user => "mysqluser" jdbc_password => "mysqlpass" jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_default_timezone =>"Asia/Shanghai" statement => "SELECT id, keyword, category_name, updated_at FROM keywords WHERE updated_at > :sql_last_value" record_last_run => "true" use_column_value => "true" tracking_column => "updated_at" tracking_column_type => "timestamp" last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_keywords" clean_run => "false" schedule => "*/3 * * * *" jdbc_paging_enabled => "true" jdbc_page_size => "50000" type => "keywords" } jdbc { jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false" jdbc_user => "mysqluser" jdbc_password => "mysqlpass" jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_default_timezone =>"Asia/Shanghai" statement => "select t.id, t.guid, t.title, t.property, t.user_id, t.user_name, t.thumbnail, t.description, t.updated_at, kg.keywords, tc.category_name from template as t left join ( select template_id, GROUP_CONCAT(keyword separator ' ') as keywords from template_keywords as tk INNER JOIN keywords as k on tk.keywords_id = k.id GROUP BY template_id ) as kg on kg.template_id = t.id inner JOIN template_category as tc on t.category_id = tc.id where t.on_market = 1 and t.updated_at > :sql_last_value" record_last_run => "true" use_column_value => "true" tracking_column => "updated_at" tracking_column_type => "timestamp" last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_template" clean_run => "false" schedule => "*/3 * * * *" jdbc_paging_enabled => "true" jdbc_page_size => "50000" type => "template" } } filter { json { source => "message" remove_field => ["message"] } } output { if[type] == "font-element" { elasticsearch { hosts => "172.30.0.106:9200" index => "logstash-ex-font_element" document_id => "%{id}" user => "elastic" password => "pass" } } if[type] == "image-element" { elasticsearch { hosts => "172.30.0.106:9200" index => "element" document_id => "image_element_%{id}" user => "elastic" password => "pass" } } if[type] == "background-element" { elasticsearch { hosts => "172.30.0.106:9200" index => "element" document_id => "background_element_%{id}" user => "elastic" password => "pass" } } if[type] == "shape-element" { elasticsearch { hosts => "172.30.0.106:9200" index => "element" document_id => "shape_element_%{id}" user => "elastic" password => "pass" } } if[type] == "keywords" { elasticsearch { hosts => "172.30.0.106:9200" index => "keywords" document_id => "keywords_%{id}" user => "elastic" password => "pass" } } if[type] == "template" { elasticsearch { hosts => "172.30.0.106:9200" index => "template" document_id => "template_%{id}" user => "elastic" password => "pass" template_overwrite => true template => "/usr/local/elk/logstash-7.4.2/config/template-es.json" } } stdout { codec => json_lines } }
2.logstash安裝logstash-input-jdbc外掛
bin/logstash-plugin install logstash-input-jdbc
3.下載mysql的驅動，注意是logstash要通過java連線mysql資料庫插入es中
驅動
https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.18/mysql-connector-java-8.0.18.jar
拷貝驅動程式到 /usr/local/elk/logstash-7.4.2/logstash-core/lib/jars 目錄下
否則會報錯：
2040 LogStash::PluginLoadingError
2040 com.mysql.jdbc.Driver not loaded. Are you sure you've included the correct jdbc driver in :jdbc_driver_library? 2038 com.mysql.jdbc.Driver not loaded. Are you sure you've included the correct jdbc driver in :jdbc_driver_library?
2038 /usr/local/elk/logstash-7.4.2/vendor/bundle/jruby/2.5.0/gems/logstash-input-jdbc-4.3.16/lib/logstash/plugin_mixins/jdbc/jdbc.rb:190:in `open_jdbc_connection'
2038 /usr/local/elk/logstash-7.4.2/vendor/bundle/jruby/2.5.0/gems/logstash-input-jdbc-4.3.16/lib/logstash/plugin_mixins/jdbc/jdbc.rb:253:in `execute_statement'
2038 /usr/local/elk/logstash-7.4.2/vendor/bundle/jruby/2.5.0/gems/logstash-input-jdbc-4.3.16/lib/logstash/inputs/jdbc.rb:309:in `execute_query'
2038 /usr/local/elk/logstash-7.4.2/vendor/bundle/jruby/2.5.0/gems/logstash-input-jdbc-4.3.16/lib/logstash/inputs/jdbc.rb:276:in `block in run'
2038 /usr/local/elk/logstash-7.4.2/vendor/bundle/jruby/2.5.0/gems/rufus-scheduler-3.0.9/lib/rufus/scheduler/jobs.rb:234:in `do_call'
4.使用systemctl管理Logstash程式
# vim /etc/systemd/system/logstash.service
[Unit] Description=logstash [Service] Type=simple User=logstash Group=logstash Environment=JAVA_HOME=/usr/local/elk/elasticsearch-7.4.2/jdk Environment=LS_HOME=/usr/local/elk/logstash-7.4.2 Environment=LS_SETTINGS_DIR=/usr/local/elk/logstash-7.4.2/config/ Environment=LS_PIDFILE=/usr/local/elk/logstash-7.4.2/logstash.pid Environment=LS_USER=logstash Environment=LS_GROUP=logstash Environment=LS_GC_LOG_FILE=/usr/local/elk/logstash-7.4.2/logs/gc.log Environment=LS_OPEN_FILES=16384 Environment=LS_NICE=19 Environment=SERVICE_NAME=logstash Environment=SERVICE_DESCRIPTION=logstash ExecStart=/usr/local/elk/logstash-7.4.2/bin/logstash "--path.settings" "/usr/local/elk/logstash-7.4.2/config/" Restart=always WorkingDirectory=/usr/local/elk/logstash-7.4.2 Nice=19 LimitNOFILE=16384 [Install] WantedBy=multi-user.target