logstash 实战操作之windows

2021-12-12

开篇原因是因为工作中使用到了elk 相关的技术，本地安装了，简单学习了下logstash。logstash是一个开源数据收集引擎，具有实时管道功能。logstash可以动态地将来自不同数据源的数据统一起来，并将数据标准化到你所选择的目的地。

第一步

下载安装包地址：https://www.elastic.co/cn/downloads/past-releases/logstash-7-5-1 windows 下载了对应的zip ,然后本地解压即可使用。

第二步

如有特殊的参数可以修改logstash-7.5.1\config文件下的logstash.yml 文件。我这里执行的默认配置。

第三步

在logstash-7.5.1\config 下创建一个配置文件，因为我这里用的mysql，命名为logstash-test-mysql.conf ，文件内容如下：

{

  stdin {
  }
  jdbc {
	#serverTimezone=UTC
	#serverTimezone=Asia/Shanghai
    jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/test?serverTimezone=GMT%2b8&useUnicode=true&characterEncoding=utf8"
    # the user we wish to excute our statement as
    jdbc_user => "root"
    jdbc_password => "1234"
    # the path to our downloaded jdbc driver  
    jdbc_driver_library => "../lib/mysql-connector-java-8.0.25.jar"
    # the name of the driver class for mysql
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    #分页设置
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"	
    #要执行的sql文件
    #statement_filepath => "/conf/test.sql"
    #statement => "select * from t_test where timestamp > date_add(:sql_last_value,INTERVAL 8 HOUR)"
    statement => "select * from t_test where status=1 and id > :sql_last_value"
    #定时配置 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
    schedule => "* * * * *"
	codec => plain { charset => "UTF-8"}
    #使用其它字段追踪，而不是用时间
    use_column_value => true
    #追踪的字段
    tracking_column => id
    record_last_run => true
    #自动生成改文件，sql_last_value 会写进该文件，执行一次后如果下次还是从该id执行不用修改，如果从头开始需要修改或删除改文件
    last_run_metadata_path => "../config/logstash_metadata"
 }
}

filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "test"
	document_id => "%{id}"
    document_type => "test"
     #该文件如下
    template =>"../config/t_test_template.json"
    template_name =>"test"
    template_overwrite =>"true"
    #user => "elastic"
    #password => "1234"
  }
  stdout {
	    # JSON格式输出
        codec => json_lines
  }
}

t_test_template.json

{
	"template" : "test",
	"mappings" : {
		"test": {
			"properties": {			 
			  "createTime": {
				"type": "date",
				"format": "yyyy-MM-dd HH:mm:ss"
			  },
			  "id": {
				"type": "keyword",			
			  },
			  "userName": {
				"type": "keyword"
			  },
			  "userId": {
				"type": "keyword"
			  },
			  "status": {
				"type": "keyword"
			  },
			  "title": {
				"type": "text",
				"analyzer": "ik_max_word",
				"search_analyzer": "ik_smart"
			  },
			  "focusPic1": {
				"type": "keyword",
				"index": false
			  }
	}
		}
	}
}

第四步

在bin 文件夹下cmd 命令黑窗口E:\elk\logstash-7.5.1\bin>输入： logstash.bat -f ../config/logstash-yydh-mysql.conf

即可执行，启动过程中，往数据库添加数据时，会通过logstash，同步到es中。前提是elasticSearch已经启动成功了。

1
2
3

[2021-08-31T19:32:28,284][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2021-08-31T19:32:28,770][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
E:/elk/logstash-7.5.1/vendor/bundle/jruby/2.5.0/gems/rufus-scheduler-3.0.9/lib/rufus/scheduler/cronline.rb:77: warning: constant ::Fixnum is deprecated