liuqianfei@Master:~/logstash-6.3.0$ ./bin/logstash -f ./script/mysql.conf Sending Logstash's logs to /home/liuqianfei/logstash-6.3.0/logs which is now configured via log4j2.properties [2018-11-21T10:55:37,328][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified [2018-11-21T10:55:39,760][INFO ][logstash.runner ] Starting Logstash {"logstash.version"=>"6.3.0"} [2018-11-21T10:55:47,803][INFO ][logstash.pipeline ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>1, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50} [2018-11-21T10:55:49,717][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}} [2018-11-21T10:55:49,802][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://localhost:9200/, :path=>"/"} [2018-11-21T10:55:50,643][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://localhost:9200/"} [2018-11-21T10:55:50,861][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>6} [2018-11-21T10:55:50,879][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>6} [2018-11-21T10:55:50,939][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil} [2018-11-21T10:55:51,040][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>60001, "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, "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"}}}}}}}} [2018-11-21T10:55:51,273][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["//localhost:9200"]} [2018-11-21T10:55:52,344][INFO ][logstash.pipeline ] Pipeline started successfully {:pipeline_id=>"main", :thread=>"#<0x349c342d run>"} [2018-11-21T10:55:52,688][INFO ][logstash.agent ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]} [2018-11-21T10:55:54,146][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600} [2018-11-21T10:55:59,368][INFO ][logstash.inputs.jdbc ] (2.011717s) SELECT * from test1_text [2018-11-21T10:57:18,337][INFO ][logstash.pipeline ] Pipeline has terminated {:pipeline_id=>"main", :thread=>"#<0x349c342d run>"}
0x349c342d>0x349c342d>
等待几分钟任务执行完毕。
es-logstash-import.jpg
这时候查询 ElasticSearch 索引，发现已经有数据了。
D:\ λ curl -X GET http://192.190.10.170:9200/_cat/indices?v health status index uuid pri rep docs.count docs.deleted store.size pri.store.size yellow open logstash-2018.11.21 SFS4m0oWSh6O30vEE3INhg 5 1 100000 0 66.mb 66.mb
D:\ λ
如果执行脚本的时候报错：
Java::JavaSql::SQLException: The server time zone value 'Öйú±ê׼ʱ¼ä' is unrecognized or represents more than one time zone.
在 JDBC 链接的 url 后要带入时区参数 ?serverTimezone=GMT%2B8。
注意 MySQL 要支持远程连接才行（如果是本地的 mysql 可以不管），不然后报拒绝访问的异常：
is not allowed to connect to this MySql server
脚本说明
最后附一个从 MySQL 定时增量导入数据的脚本和参数说明，仅供参考。
#-----------------------------------start----------------------------------- #输入部分 input { jdbc { #连接MySQL驱动，需要自己下载 jdbc_driver_library => "/es/mysql-connector-java-5.1.31.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://10.112.29.30:3306/mstore" #连接数据库账号信息 jdbc_user => "MySQL_admin" jdbc_password => "password" #分页 jdbc_paging_enabled => true #分页大小 jdbc_page_size => 100000 #流式获取数据，每次取10000. jdbc_fetch_size => 10000 #Maximum number of times to try connecting to database connection_retry_attempts => 3 #Number of seconds to sleep between connection attempts connection_retry_attempts_wait_time => 1 #Connection pool configuration. The amount of seconds to wait to acquire a connection before raising a PoolTimeoutError (default 5) jdbc_pool_timeout => 5 #Whether to force the lowercasing of identifier fields lowercase_column_names => true #Whether to save state or not in last_run_metadata_path #保存上次运行记录，增量提取数据时使用 record_last_run = > true #"* * * * *"为每分钟执行一次 schedule => "* * * * *" #Use an incremental column value rather than a timestamp use_column_value => true #sql_last_value #The value used to calculate which rows to query. Before any query is run, this is set to Thursday, 1 January 1970, or 0 if use_column_value is true and tracking_column is set. It is updated accordingly after subsequent queries are run. tracking_column => "id" #查询语句 statement => "SELECT id,package_name,name,sub_name,editor_comment,high_quality,sub_category,tag,update_time FROM tbl_app WHERE id > :sql_last_value" } }
