下载

https://artifacts.elastic.co/downloads/logstash/logstash-6.7.1.tar.gz

安装

上传下载过后的压缩包到自己的服务器目录

# 进入目录
cd /opt/software

# 解压到指定目录
tar -zxvf logstash-6.7.1.tar.gz -C /usr/local

配置

新建配置文件

cp logstash-sample.conf logstash-test.conf

# 进入logstash数据目录
cd /usr/local/logstash-6.7.1/data
# 新建结果保存目录
mkdir metadata
# 新建抽取脚本目录
mkdir sql

修改配置文件

jdbc_driver_library： 数据库驱动存放位置

statement_filepath：抽取脚本存放位置

last_run_metadata_path：运行结果保存位置

部分配置文件

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
    jdbc {
    	#驱动地址
    	jdbc_driver_library => "/home/xis/ojdbc8-12.2.0.1.jar"
	    # jdbc_driver_class => "com.mysql.jdbc.Driver"
	    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
	    jdbc_connection_string => "jdbc:oracle:thin:@192.168.40.104:1521:hczz"
	    jdbc_user => zdkz
	    jdbc_password => zdkz
	    #仅提取自上次运行以来更改的数据一般使用自增id或者使用update_time 最后更新时间
	    #sql 语句文件，也可以直接写SQL，如statement => "select * from table1"
	    #statement => "select * from tb_yw_shxx where del = '0' and modify_date > :sql_last_value"
        statement_filepath => "/usr/local/logstash-6.7.1/data/sql/shxx.sql"
        #是否使用column 作为增量字段
	    use_column_value => true
	    #增量字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
	    tracking_column_type => "timestamp"
	    #以id做为增量值，用于增量同步，需是数据库字段
	    tracking_column => "modify_date"
	    #是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中
	    record_last_run => true
	    #上面运行结果的保存位置
	    last_run_metadata_path => "/usr/local/logstash-6.7.1/data/metadata/shxx"	    
		# 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		lowercase_column_names => true       	    
	    #是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false
	  	clean_run => false
	  	#定时设置，corn表达式
      	schedule => "* * * * *"
      	#ES中type
      	type => "shxx"
	}	
    jdbc {
    	#驱动地址
    	jdbc_driver_library => "/home/xis/ojdbc8-12.2.0.1.jar"
	    # jdbc_driver_class => "com.mysql.jdbc.Driver"
	    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
	    jdbc_connection_string => "jdbc:oracle:thin:@192.168.40.104:1521:hczz"
	    jdbc_user => zdkz
	    jdbc_password => zdkz
	    #仅提取自上次运行以来更改的数据一般使用自增id或者使用update_time 最后更新时间
	    #sql 语句文件，也可以直接写SQL，如statement => "select * from table1"
	    #statement => "select * from tb_yw_zdxx where del = '0' and modify_date > :sql_last_value order"
        statement_filepath => "/usr/local/logstash-6.7.1/data/sql/zdxx.sql"
        #是否使用column 作为增量字段
	    use_column_value => true
	    #增量字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
	    tracking_column_type => "timestamp"
	    #以id做为增量值，用于增量同步，需是数据库字段
	    tracking_column => "modify_date"
	    #是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中
	    record_last_run => true
	    #上面运行结果的保存位置
	    last_run_metadata_path => "/usr/local/logstash-6.7.1/data/metadata/zdxx"	    
		# 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		lowercase_column_names => true       	    
	    #是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false
	  	clean_run => false
	  	#定时设置，corn表达式
      	schedule => "* * * * *"
		#ES中type
      	type => "zdxx"      	
	}				    	
}

filter {
    # 解决默认时区与当前时区相差8小时问题
	ruby { 
		code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)" 
	}
	ruby {
		code => "event.set('@timestamp',event.get('timestamp'))"
	}
	mutate {
		remove_field => ["timestamp"]
	}
}

output {
	if[type] == "shxx"{
		elasticsearch {
			#ES ip:port
			hosts => ["localhost:9200"]
			#ES索引名
			index => "tb_yw_shxx"
			document_type => "doc"
			document_id => "%{id}"
		}
	}
	if[type] == "zdxx"{
		elasticsearch {
			#ES ip:port
			hosts => ["localhost:9200"]
			#ES索引名
			index => "tb_yw_zdxx"
			document_type => "doc"
			document_id => "%{id}"	
		}
	}
}

启动

# 进入logstash安装目录
cd /usr/local/logstash-6.7.1/bin

# 启动
logstash -f ../config/logstash-test.conf

后台启动

nohup logstash -f ../config/logstash-test.conf

停止

查看进程号 ps -ef|grep logstash

杀掉进程 kill -9 #logstash进程号