logstash6-mysql同步数据到es

解压

unzip logstash-6.6.0.zip

前台启动

/mnt/escache/elasticsearch-6.6.0/logstash-6.6.0/bin/logstash -f /mnt/escache/elasticsearch-6.6.0/logstash-6.6.0/jdbc_questions.config

//后台启动

nohup /mnt/escache/elasticsearch-6.6.0/logstash-6.6.0/bin/logstash -f /mnt/escache/elasticsearch-6.6.0/logstash-6.6.0/jdbc_questions.config &

/mnt/soft/logstash-6.6.0/bin/logstash -f /mnt/soft/logstash-6.6.0/jdbc_questions.config

查看进程

ps -ef |grep logstash

加载 etc文件夹下所有 *.conf 的文本文件，然后在自己内存里拼接成一个完整的大配置文件

# ./bin/logstash -f config/

后台运行

# nohup ./bin/logstash -f config/ &

3.4 停止

查找进程 ID

# ps -ef |grep logstash


或查看 9600 端口

# netstat -ntlp|grep java

相关配置信息

input {
    stdin {
    }
    jdbc {
      // mysql相关jdbc配置
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/your_mysql_database"
      jdbc_user => "mysql_user"
      jdbc_password => "mysql_password"
 
      // jdbc连接mysql驱动的文件目录，可去官网下载:https://dev.mysql.com/downloads/connector/j/
      jdbc_driver_library => "/opt/logstash/conf/mysql-connector-java/mysql-connector-java-5.1.44-bin.jar"
      # the name of the driver class for mysql
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"

      // mysql文件, 也可以直接写SQL语句在此处，如下：
      // statement => "SELECT * from Table_test;"
      statement_filepath => "/opt/logstash/conf/jdbc.sql"

      // 这里类似crontab,可以定制定时操作，比如每10分钟执行一次同步(分 时 天 月 年)
      //schedule => "* * * * *"     1分钟执行一次
      schedule => "*/10 * * * *"
      type => "jdbc"

      // 是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
      record_last_run => "true"

      // 是否需要记录某个column 的值,如果record_last_run为真,可以自定义我们需要 track 的 column 名称，此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.
      use_column_value => "true"

      // 如果 use_column_value 为真,需配置此参数. track 的数据库 column 名,该 column 必须是递增的. 一般是mysql主键
      tracking_column => "autoid"

      last_run_metadata_path => "/opt/logstash/conf/last_id"

      // 是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
      clean_run => "false"

      //是否将 字段(column) 名称转小写
      lowercase_column_names => "false"
    }
}

// 此处我不做过滤处理,如果需要，也可参考elk安装那篇
filter {}

output {
    // 输出到elasticsearch的配置
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "jdbc"

        // 将"_id"的值设为mysql的autoid字段
        document_id => "%{autoid}"
         //会覆盖

        template_overwrite => true
    }

    // 这里输出调试，正式运行时可以注释掉
    stdout {
        codec => json_lines
    }
}

一次同步多个表

核心配置文件

input {
  stdin {
  }

  jdbc {
  type => "cxx_article_info"
  # mysql jdbc connection string to our backup databse 后面的test对应mysql中的test数据库
  jdbc_connection_string => "jdbc:mysql://110.10.15.37:3306/cxxwb"
  # the user we wish to excute our statement as
  jdbc_user => "root"
  jdbc_password => "xxxxx"

  record_last_run => "true"
  use_column_value => "true"
  tracking_column => "id"
  last_run_metadata_path => "/opt/logstash/bin/logstash_xxy/cxx_info"
  clean_run => "false"

  # the path to our downloaded jdbc driver
  jdbc_driver_library => "/opt/elasticsearch/lib/mysql-connector-java-5.1.38.jar"
  # the name of the driver class for mysql
  jdbc_driver_class => "com.mysql.jdbc.Driver"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "500"
  statement => "select * from cxx_article_info where id > :sql_last_value"
#定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
  schedule => "* * * * *"
#设定ES索引类型
  }

  jdbc {
  type => "cxx_user"
  # mysql jdbc connection string to our backup databse 后面的test对应mysql中的test数据库
  jdbc_connection_string => "jdbc:mysql://110.10.15.37:3306/cxxwb"
  # the user we wish to excute our statement as
  jdbc_user => "root"
  jdbc_password => "xxxxxx"

  record_last_run => "true"
  use_column_value => "true"
  tracking_column => "id"
  last_run_metadata_path => "/opt/logstash/bin/logstash_xxy/cxx_user_info"
  clean_run => "false"

  # the path to our downloaded jdbc driver
  jdbc_driver_library => "/opt/elasticsearch/lib/mysql-connector-java-5.1.38.jar"
  # the name of the driver class for mysql
  jdbc_driver_class => "com.mysql.jdbc.Driver"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "500"
  statement => "select * from cxx_user_info where id > :sql_last_value"
#以下对应着要执行的sql的绝对路径。
#statement_filepath => "/opt/logstash/bin/logstash_mysql2es/department.sql"
#定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
schedule => "* * * * *"
#设定ES索引类型
  }

}

filter {
mutate {
  convert => [ "publish_time", "string" ]
 }

date {
  timezone => "Europe/Berlin"
  match => ["publish_time" , "ISO8601", "yyyy-MM-dd HH:mm:ss"]
}
#date {
 # match => [ "publish_time", "yyyy-MM-dd HH:mm:ss,SSS" ]
  # remove_field => [ "publish_time" ]
  # }
json {
  source => "message"
  remove_field => ["message"]
  }
}

output {

if [type]=="cxxarticle_info" {
  elasticsearch {
#ESIP地址与端口
  hosts => "10.100.11.231:9200"
#ES索引名称（自己定义的）
  index => "cxx_info_index"
#自增ID编号
 # document_id => "%{id}"
  }
}

if [type]=="cxx_user" {
  elasticsearch {
#ESIP地址与端口
  hosts => "10.100.11.231:9200"
#ES索引名称（自己定义的）
  index => "cxx_user_index"
#自增ID编号
 # document_id => "%{id}"
  }
}

}

定时设置

quartz Cron

CronTrigger

CronTriggers往往比SimpleTrigger更有用，如果您需要基于日历的概念，而非SimpleTrigger完全指定的时间间隔，复发的发射工作的时间表。
CronTrigger，你可以指定触发的时间表如“每星期五中午”，或“每个工作日9:30时”，甚至“每5分钟一班9:00和10:00逢星期一上午，星期三星期五“。
即便如此，SimpleTrigger一样，CronTrigger拥有的startTime指定的时间表时生效，指定的时间表时，应停止（可选）结束时间。

Cron Expressions

cron的表达式被用来配置CronTrigger实例。 cron的表达式是字符串，实际上是由七子表达式，描述个别细节的时间表。这些子表达式是分开的空白，代表：

1. 1.        Seconds
2. 2.        Minutes
3. 3.        Hours
4. 4.        Day-of-Month
5. 5.        Month
6. 6.        Day-of-Week
7. 7.        Year (可选字段)

例  "0 0 12 ? * WED" 在每星期三下午12:00 执行,

个别子表达式可以包含范围, 例如，在前面的例子里("WED")可以替换成 "MON-FRI", "MON, WED, FRI"甚至"MON-WED,SAT".

“*” 代表整个时间段.

每一个字段都有一套可以指定有效值，如

Seconds (秒)         ：可以用数字0－59 表示，

Minutes(分)          ：可以用数字0－59 表示，

Hours(时)             ：可以用数字0-23表示,

Day-of-Month(天) ：可以用数字1-31 中的任一一个值，但要注意一些特别的月份

Month(月)            ：可以用0-11 或用字符串  “JAN, FEB, MAR, APR, MAY, JUN, JUL, AUG, SEP, OCT, NOV and DEC” 表示

Day-of-Week(每周)：可以用数字1-7表示（1 ＝ 星期日）或用字符口串“SUN, MON, TUE, WED, THU, FRI and SAT”表示

“/”：为特别单位，表示为“每”如“0/15”表示每隔15分钟执行一次,“0”表示为从“0”分开始, “3/20”表示表示每隔20分钟执行一次，“3”表示从第3分钟开始执行

“?”：表示每月的某一天，或第周的某一天

“L”：用于每月，或每周，表示为每月的最后一天，或每个月的最后星期几如“6L”表示“每月的最后一个星期五”

“W”：表示为最近工作日，如“15W”放在每月（day-of-month）字段上表示为“到本月15日最近的工作日”

““#”：是用来指定“的”每月第n个工作日,例 在每周（day-of-week）这个字段中内容为"6#3" or "FRI#3" 则表示“每月第三个星期五”

1）Cron表达式的格式：秒 分 时 日 月 周 年(可选)。

               字段名                 允许的值                        允许的特殊字符  
               秒                         0-59                               , - * /  
               分                         0-59                               , - * /  
               小时                     0-23                               , - * /  
               日                         1-31                               , - * ? / L W C  
               月                         1-12 or JAN-DEC         , - * /  
               周几                     1-7 or SUN-SAT           , - * ? / L C #  
               年 (可选字段)     empty, 1970-2099      , - * /

               “?”字符：表示不确定的值

               “,”字符：指定数个值

               “-”字符：指定一个值的范围

               “/”字符：指定一个值的增加幅度。n/m表示从n开始，每次增加m

               “L”字符：用在日表示一个月中的最后一天，用在周表示该月最后一个星期X

               “W”字符：指定离给定日期最近的工作日(周一到周五)

               “#”字符：表示该月第几个周X。6#3表示该月第3个周五

         2）Cron表达式范例：

                 每隔5秒执行一次：*/5 * * * * ?

                 每隔1分钟执行一次：0 */1 * * * ?

                 每天23点执行一次：0 0 23 * * ?

                 每天凌晨1点执行一次：0 0 1 * * ?

                 每月1号凌晨1点执行一次：0 0 1 1 * ?

                 每月最后一天23点执行一次：0 0 23 L * ?

                 每周星期天凌晨1点实行一次：0 0 1 ? * L

                 在26分、29分、33分执行一次：0 26,29,33 * * * ?

                 每天的0点、13点、18点、21点都执行一次：0 0 0,13,18,21 * * ?