首先声明logstash的版本是6.x，而非7.x，若7.0则提示：
You are using a deprecated config setting "document_type" set in elasticsearch. Deprecated settings will continue to work, but are scheduled for removal from logstash in the future. Document types are being deprecated in Elasticsearch 6.0, and removed entirely in 7.0.
1、logstash的下载
官网 http://www.elastic.co/cn/products/logstash
ELK 都是 Elastic 公司的产品，所有产品的下载地址： https://www.elastic.co/cn/downloads/，直接下载安装包可以到 https://www.elastic.co/cn/downloads/logstash，老版本在 https://www.elastic.co/cn/downloads/past-releases#logstash
releasenote：https://www.elastic.co/guide/en/logstash/6.8/releasenotes.html
es的下载 http://www.elastic.co/downloads/elasticsearch 下载，老版本的归档在 https://www.elastic.co/downloads/past-releases 页面选择下载。
2、下载后直接解压 logstash，进入到logstash bin目录
cd logstash-6.4.0/bin
现在使用新版本的Logstash 已经不需要单独安装
logstash-input-jdbc
插件，
logstash-6.1.1
以后已经默认支持
logstash-input-jdbc
插件，不需要再单独安装了。
所以可以执行
./logstash -e
但是如果有提示错误 ERROR: Pipelines YAML file is empty. config/pipelines.yml，需要先配置下pipelines.yml，如下：
这样配置文件为在指定目录下的后缀为conf的文件
如果是老版本需要安装jdbc的入插件和elasticsearch的出插件：logstash-input-jdbc、logstash-output-elasticsearch
3：安装logstash-jdbc和logstash-output-elasticsearch
（1）./logstash-plugin install logstash-input-jdbc
（2）./logstash-plugin install logstash-output-elasticsearch
4：编写配置文件（jdbc.sql和jdbc.conf，建议在bin目录下vim jdbc.conf)
（1）：首先在bin目录下新建一个mysql目录,里面包含jdbc.conf,jdbc.sql文件，也可以只有jdbc.conf文件，具体看里面配置就懂了,然后加入mysql的驱动
jdbc.conf内容：
input {
stdin {
}
jdbc {
# 连接的数据库地址和哪一个数据库，指定编码格式，禁用SSL协议，设定自动重连
jdbc_connection_string => "jdbc:mysql://数据库地址:端口号/数据库名?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
# 你的账户密码
jdbc_user => "账号"
jdbc_password => "密码"
# 连接数据库的驱动包，建议使用绝对地址
jdbc_driver_library => "mysql/mysql-connector-java-5.1.45-bin.jar"
# 这是不用动就好
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
#处理中文乱码问题
codec => plain { charset => "UTF-8"}
#使用其它字段追踪，而不是用时间
use_column_value => true
#追踪的字段
tracking_column => id
record_last_run => true
#上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值
last_run_metadata_path => "mysql/station_parameter.txt"
jdbc_default_timezone => "Asia/Shanghai"
#此处也可以直接写sql，而不写到文件中，key变为statement
statement_filepath => "mysql/jdbc.sql"
#是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
clean_run => false
# 这是控制定时的，重复执行导入任务的时间间隔，第一位是分钟
schedule => "* * * * *"
type => "jdbc"
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
}
output {
elasticsearch {
# 要导入到的Elasticsearch所在的主机
hosts => "192.168.10.210:9200"
# 要导入到的Elasticsearch的索引的名称
index => "db_case"
# 类型名称（类似数据库表名）
document_type => "table_caseitem"
# 主键名称（类似数据库主键）
document_id => "%{id}"
}
stdout {
# JSON格式输出
codec => json_lines
}
}
jdbc.sql里面就直接写sql语句就行了，
如果全量同步sql语句就不要写testid了；增量同步就加上where条件 where testid > :sql_last_value
5、开始导入(需要用root账户启动)
./logstash -f mysql/jdbc.conf
6、多配置文件
我们知道在启动logstash的时候，只要加上
-f /you_path_to_config_file
就可以加载配置文件了，但是如果需要多个配置文件，应该怎么写呢？
只需要
-f /you_path_to_config_directory
就可以了。简单说，就是在
-f
后面加上目录就可以。
注意：目录后面不能加 * 号
7、input、filter和output
logstash多个配置文件里的input、filter、output是否相互独立呢？答案是否定的。
怎么办呢？
建议使用
tags
或者
type
这两个特殊字段，即在读取文件的时候，添加标识符在
tags
中或者定义
type
变量。
input{
file{
path=>[
"/home/husen/log/sys.log"
]
type => "from_sys"
#tags => ["from_sys"]
}
}
input{
file{
path=>[
"/home/husen/log/error.log"
]
type => "from_error"
#tags => ["from_sys"]
}
}
if [type] == "from_sys"{
#if "from_sys" in [tags]
elasticsearch {
action => "index"
hosts => "localhost:9200"
index => "from_sys_log"
codec => "json"
}
}
if [type] == "from_error"{
#if "from_error" in [tags]
elasticsearch {
action => "index"
hosts => "localhost:9200"
index => "from_error_log"
codec => "json"
}
}
//特别地，如果要针对不同的类型日志用不同filter来grok解析，
//也可以通过类似的方法判断
若有的字段需要转变成date类型的字段，只需在filter中加上以下部分：
filter {
date {
match => [ "created_at", "yyyy-MM-dd HH:mm:ss" ]
}
}
这样便可讲created_at字段 格式为yyyy-MM-dd HH:mm:ss的变为date类型，详情见 https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html#plugins-filters-date-match