[关闭]
@xiaoxiaowang 2019-06-25T15:07:11.000000Z 字数 5499 阅读 1355

Logstash 简易教程

Logstash 数据同步

建议在使用logstash之前先想清楚自己的需求是什么，从哪种数据源同步到哪里，需要经过怎么样的处理。因为logstash版本迭代较快，每个版本的插件都有点区别，比如filter中的http插件在6.6版本以后才有；output到现在(7.1)都没有jdbc的插件，然而你如果想使用output的jdbc插件就需要自己去安装热心人自己写的插件(logstash-output-jdbc),不幸的是，该作者指出没有很多的时间去维护此插件，不能保证6.3以后用户的正常使用。
也就是说，如果你想用output的jdbc，你就必须使用6.3以下(最好5.x)的版本，如果你想用官方filter的http插件，你就得用6.5以上的版本。话说logstash为什么不把jdbc加入到output中去呢。。。

另外：最好在linux下进行调试或使用，因为用windows会出现恶心的乱码问题。。。。

安装

如果目标数据源的type为jdbc，则建议安装logstash-5.4.1或6.x一下的版本，因为logstash-output-jdbc只在6.x以下有效，本教程对应的版本为logstash-5.4.1

建议直接下载zip包进行使用，解压完之后的目录如下

  1. .
  2. ├── bin  # 二进制脚本，包括用于启动Logstash的logstash和用于安装插件的logstash-plugin
  3. ├── config # Configuration files, including logstash.yml and jvm.options
  4. ├── CONTRIBUTORS
  5. ├── data
  6. ├── Gemfile
  7. ├── Gemfile.jruby-2.3.lock
  8. ├── lib
  9. ├── LICENSE
  10. ├── logs # 日志文件
  11. ├── logstash-core
  12. ├── logstash-core-plugin-api
  13. ├── modules
  14. ├── NOTICE.TXT
  15. ├── tools
  16. └── vendor

https://www.elastic.co/guide/en/logstash/7.1/dir-layout.html

可以在对应目录下，执行如下命令来验证是否安装成功

  1. bin/logstash -e 'input { stdin { } } output { stdout {} }'

等待启动成功，直接输入“hello world”
会有如下显示

  1. {
  2.       "@version" => "1",
  3.     "@timestamp" => 2019-06-20T06:46:48.310Z,
  4.        "message" => "hello world",
  5.           "host" => "localhost.localdomain"
  6. }

则安装成功

语法

区段

Logstash 设计了自己的 DSL —— 有点像 Puppet 的 DSL，或许因为都是用 Ruby 语言写的吧 —— 包括有区域，注释，数据类型(布尔值，字符串，数值，数组，哈希)，条件判断，字段引用等。

Logstash 用 {} 来定义区域，区域内可以包括插件区域定义，你可以在一个区域内定义多个插件。插件区域内则可以定义键值对设置。示例如下：

  1. input {}
  2. filter {}
  3. output {}

数据类型

类型 示例
bool debug => true
bytes my_bytes => "113" # 113 bytes
string host => "hostname"
number port => 214
array match =>[ "/var/log/messages", "/var/log/*.log" ]
hash options => {key1 => "value1",key2 => "value2" }

条件判断

名称 符号
等于 ==
不等于 !=
小于 <
大于 >
小于等于 <=
大于等于
=
匹配正则 =~
不匹配正则 !~
包含 in
不包含 not in
与 and
或 or
非与 nand
非或 xor
复合表达式 ()
取反符合 !()

示例

读取日志文件

https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html#plugins-inputs-file-type
各参数介绍与赋值

全量同步一个文件的所有内容，并且进行监听

  1. input {
  2.     file {
  3.         path => ["/usr/local/logstash/logstash-tutorial-dataset"]
  4.         type => "file_monitor"
  5.         tags => ["有用的","标识用的"]
  6.         start_position => "beginning"
  7.     }
  9. }
  10. output {
  11.    stdout {}
  12. }

监听后不能打开编辑再保存必须使用echo >> 对文件进行追加内容，可以看到

  1. {
  2.        "message" => "2019年6月20日16:57:52",
  3.     "@timestamp" => 2019-06-20T08:57:55.624Z,
  4.           "tags" => [
  5.         [0] "有用的",
  6.         [1] "标识用的"
  7.     ],
  8.           "host" => "localhost.localdomain",
  9.           "path" => "/usr/local/logstash/logstash-tutorial-dataset",
  10.       "@version" => "1",
  11.           "type" => "file_monitor"
  12. }

读取mysql

  1. input {
  2.     jdbc {
  3.         jdbc_driver_library => "/usr/local/logstash/mysql/mysql-connector-java-5.1.47.jar"
  4.         jdbc_driver_class => "com.mysql.jdbc.Driver"
  5.         jdbc_connection_string => "jdbc:mysql://xxx:xxx/data"
  6.         jdbc_user => "xxx"
  7.         jdbc_password => "xxx"
  8.         parameters => { "id" => 666 }
  9.         statement => "SELECT * FROM xxx_t_job_function_net_bak20140828 WHERE id > :id"
  10.     }
  11. }
  12. output {
  13.     stdout {}
  14. }

写mysql

官方没有提供写mysql的插件，所以需要自己下载安装
https://github.com/theangryangel/logstash-output-jdbc

./logstash-plugin install logstash-output-jdbc

可惜的是，目前插件只支持6.3以前，当前logstash最新版本为7.1.1，
测试用的是logstash-5.4.1 较稳定

  1. input {
  2.     jdbc {
  3.         jdbc_driver_library => "/usr/local/logstash/mysql/mysql-connector-java-5.1.47.jar"
  4.         jdbc_driver_class => "com.mysql.jdbc.Driver"
  5.         jdbc_connection_string => "jdbc:mysql://120.55.168.18:8840/data"
  6.         jdbc_user => "crawler"
  7.         jdbc_password => "crawler9988@ed"
  8.         parameters => { "id" => 666 }
  9.         statement => "SELECT * FROM xxx_t_job_function_net_bak20140828 WHERE id > :id"
  10.     }
  11. }
  12. output {
  13.     jdbc {
  14.         driver_jar_path => "D:\repo\mysql\mysql-connector-java\5.1.40\mysql-connector-java-5.1.40.jar"
  15.         driver_class => "com.mysql.jdbc.Driver"
  16.         connection_string => "jdbc:mysql://sss:8840/testcase"
  17.         username => "sss"
  18.         password => "csssd"
  19.         statement => ["INSERT INTO job_function_20190621 ( code_val, name_val, level_val, source_name, version ) VALUES (?,?,?,?,?)","code","name","level","source_name","current_version"]
  20.     }
  21.     stdout {}
  22. }

input

output

filter

以上三个链接为官方文档，基本上涵盖了可能会用到的数据源头与目标，如果没有的话，可以去github上搜索一下一般都会有人自己开发插件如logstash-output-jdbc

完整的示例

  1. input {
  2.     jdbc {
  3.         jdbc_driver_library => "D:\repo\mysql\mysql-connector-java\5.1.40\mysql-connector-java-5.1.40.jar"    
  4.         jdbc_driver_class => "com.mysql.jdbc.Driver"
  5.         jdbc_connection_string => "jdbc:mysql://xxx:xxx/data"
  6.         jdbc_user => "xxx"
  7.         jdbc_password => "xxxx"
  8.         statement => "SELECT * FROM xxx_t_job_function_net_bak20140828 WHERE id > :id"
  9.         parameters => { "id" => 666 }
  10.     }
  11. }
  12. filter {
  13.     grok {
  14.         match => {"@timestamp" => "%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day}" }
  15.         add_field => { "current_version" => "%{year}%{month}%{day}"}
  16.     }
  17.     jdbc_streaming {
  18.         # 加了会错误，可能引用了上面的input jdbc_driver_library => "D:\repo\mysql\mysql-connector-java\5.1.40\mysql-connector-java-5.1.40.jar"
  19.         jdbc_driver_class => "com.mysql.jdbc.Driver"
  20.         jdbc_connection_string => "jdbc:mysql://xxx:3306/xxx"
  21.         jdbc_user => "xxx"
  22.         jdbc_password => "xxx"
  23.         statement => "SELECT location_name_cn FROM dict_location WHERE location_code = :codeParam"
  24.         parameters => { "codeParam" => "code"}
  25.         target => "code"
  26.     }
  27.     if [code] and [code][0] and ("location_name_cn" in [code][0]) {
  28.         ruby {
  29.             code => "
  30.             r = ''
  31.             event.get('code').each do |variable|
  32.                # puts variable['location_name_cn']
  33.                r = r + variable['location_name_cn'] + ';'
  34.             end 
  35.             event.set('code',r)
  36.             "
  37.         }
  38.     } else {
  39.         mutate {
  40.             replace => { "code" => ""}
  41.         }
  42.     }
  43. }
  44. output {
  45.     stdout {
  46.         codec => rubydebug{}        
  47.     }
  48.     jdbc {
  49.         driver_jar_path => "D:\repo\mysql\mysql-connector-java\5.1.40\mysql-connector-java-5.1.40.jar"
  50.         driver_class => "com.mysql.jdbc.Driver"
  51.         connection_string => "jdbc:mysql://xxx:8840/xxx"
  52.         username => "xxx"
  53.         password => "xxx"
  54.         statement => ["INSERT INTO job_function_20190621 ( code_val, name_val, level_val, source_name, version ) VALUES (?,?,?,?,?)","code","name","level","source_name","current_version"]
  55.     }
  56. }

最佳实践

1、整理并调试好读取数据与插入数据的脚本或参数；

2、一般字符串的操作官方给的插件中就可以解决如Date、json、mutate等

3、遇到比较复杂的操作可以借助ruby，在编译器中调试好，直接复制过来，可以实时调试；

4、grok为提取日志的核心操作组件

5、注意变量的引用，内置参数为@timestamp

字符串中引用 "the longitude is %{[geoip][location][0]}"
作为判断的时候可以[filed]使用

https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注