1 MySql数据到Elasticsearch

1.1 下载logstash

官网
https://www.elastic.co/cn/logstash/

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

1.2 解压logstash

tar -zxvf logstash-6.6.0.tar.gz

1.3 在logstash 目录创建 mysql 文件夹

[root@bigdata01 logstash-6.6.0]# mkdir mysql

1.4 将 mysql 驱动文件和数据库查询文件 放进mysql中

1.5 在config 目录下创建 mysqltoes.conf 文件

1.6 mysqltoes.conf 配置

input 
  
  # 多张表的同步只需要设置多个jdbc的模块就行了
  jdbc 
      # mysql 数据库链接,shop为数据库名
      jdbc_connection_string => "jdbc:mysql://ip:3306/mall?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true"
      # 用户名和密码
      jdbc_user => "root"
      jdbc_password => "root"

      # 驱动
      jdbc_driver_library => "/usr/local/logstash-6.6.0/mysql/mysql-connector-java-8.0.16.jar"

      # 驱动类名
      jdbc_driver_class => "com.mysql.jdbc.Driver"

      #是否分页
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"

      #直接执行sql语句
      # statement =>"select * from t_item"
      # 执行的sql 文件路径+名称
      statement_filepath => "/usr/local/logstash-6.6.0/mysql/item.sql"
      
      # 默认列名转换为小写
      lowercase_column_names => "false"

      #设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
      schedule => "* * * * *"

      # 索引类型
      #type => "jdbc"
    




output 
  elasticsearch 
        #es的ip和端口
        hosts => ["http://ip:9200"]
        #ES索引名称（自己定义的）
            index => "mall"
        #文档类型
        document_type => "mall_item"
        #设置数据的id为数据库中的字段
        document_id => "%iteId"
    
    stdout 
        codec => json_lines
    

1.7 启动 logstash

前台启动：

[root@bigdata01 bin]# ./logstash -f ../config/mysqltoes.conf

后台启动：

[root@bigdata01 bin]# nohup  ./logstash -f ../config/mysqltoes.conf >logstash.log &

2 配置语法讲解

logstash使用 来定义配置区域，区域内又可以包含其插件的区域配置。

# 最基本的配置文件定义，必须包含input 和 output。

input
    stdin 



output
    stdout
        codec=>rubydebug
    



# 如果需要对数据进操作，则需要加上filter段

input
    stdin 



filter
  # 里面可以包含各种数据处理的插件，如文本格式处理 grok、键值定义 kv、字段添加、
  # geoip 获取地理位置信息等等... 



output
    stdout
        codec=>rubydebug
    



# 可以定义多个输入源与多个输出位置

input
    stdin 
    
    file
        path => ["/var/log/message"]
        type => "system"
        start_position => "beginning"
    



output
    stdout
        codec=>rubydebug
    
    
     
    file 
        path => "/var/datalog/mysystem.log.gz"
        gzip => true
    
    

3 启动方式

# 通过手动指定配置文件启动

/bin/logstash -f /etc/logstash/conf.d/nginx_logstash.conf

# 以daemon方式运行，则在指令后面加一个 & 符号

/bin/logstash -f /etc/logstash/conf.d/nginx_logstash.conf &


# 如果是通过rpm包安装的logstash则可以使用自带的脚本启动

/etc/init.d/logstash start 

# 通过这种方式启动，logstash会自动加载 /etc/logstash/conf.d/ 下的配置文件

4 filebeat基本讲解

filebeat是基于原先 logstash-forwarder 的源码开发而来，无需JAVA环境，运行起来更轻便，无疑是业务服务器端的日志收集工具。

配 置

# 配置文件路径 "/etc/filebeat/filebeat.yml"
# 一个配置文件可以包含多个prospectors，一个prospectors可以包含多个path。


filebeat:
  # List of prospectors to fetch data.
  prospectors:
    # Each - is a prospector. Below are the prospector specific configurations
    -
      paths:
        - /var/log/messages
        
      input_type: log

      document_type: messages

    -
      paths:
        - /alidata/log/nginx/access/access.log.json

      input_type: log

      document_type: nginxacclog

    -
      paths:
        - /alidata/www/storage/logs/laravel.log

      input_type: log

      document_type: larlog

    -
      paths:
        - /alidata/www/500_error/500_error.log

      input_type: log

      document_type: error500

    -
      paths:
        - /alidata/www/deposit/deposit.log

      input_type: log

      document_type: deposit


    -
      paths:
        - /alidata/www/call_error.log

      input_type: log

      document_type: call_error

    -
      paths:
        - /alidata/www/weixin_deposit.log

      input_type: log

      document_type: weixin_deposit


    -
      paths:
        - /alidata/log/php/php-fpm.log.slow

      input_type: log

      document_type: phpslowlog

      # 多行处理
      multiline:
          pattern: '^[[:space:]]'
          negate: true
          match: after


    # Additional prospector

registry_file: /var/lib/filebeat/registry


############################# Libbeat Config ##################################
# Base config file used by all other beats for using libbeat features

############################# Output ##########################################

# 输出数据到 redis 

output:


  redis:

    host: "10.122.52.129"
    port: 6379
    password: "123456"


# 输出数据到 logstash ，一般两者选用其一
 
  logstash:
    hosts: ["10.160.8.221:5044"]


############################# Shipper #########################################

shipper:

# 打上服务器tag

  name: "host_2"
  
  
############################# Logging #########################################  
  
logging:  
  
  
  files:
  

    rotateeverybytes: 10485760 # = 10MB

filebeat主要配置就是这个配置文件了，设定好之后启动服务就会自动从源拉取数据发送到指定位置，当数据为普通行数据时，filebeat会自动为其添加字段信息，其中一项字段 @timestamp 为filebeat读取到这条数据的时间，默认格式为UTC时间，比中国大陆时间早8小时。

如果数据为json格式，而数据中已包含@timestamp字段，filebeat处理时会把@timestamp字段值替换为filebeat读取到该行数据的当前UTC时间。

5 实战运用

5.1 业务到redis到es之间迁移

nginx 日志格式配置

       log_format json '"@timestamp":"$time_iso8601",'
                 '"slbip":"$remote_addr",'
                 '"clientip":"$http_x_forwarded_for",'
                 '"serverip":"$server_addr",'
                 '"size":$body_bytes_sent,'
                 '"responsetime":$request_time,'
                 '"domain":"$host",'
                 '"method":"$request_method",'
                 '"requesturi":"$request_uri",'
                 '"url":"$uri",'
                 '"appversion":"$HTTP_APP_VERSION",'
                 '"referer":"$http_referer",'
                 '"agent":"$http_user_agent",'
                 '"status":"$status"';

filebeat 配置

filebeat:
  # List of prospectors to fetch data.
  prospectors:
    # Each - is a prospector. Below are the prospector specific configurations

    -
      paths:
        - /alidata/log/nginx/access/access.log.json

      input_type: log

      document_type: nginxacclog
      

############################# Output ##########################################

output:            
 
  logstash:
    hosts: ["10.160.8.221:5044"]

# 其他部分配置省略。

logstash 配置 （此处logstash用于接收filebeat的数据，然后转存redis）

input 
    beats 
    port => 5044
    codec => "json"





filter 
    if [type] == "nginxacclog" 
    geoip 
        source => "clientip"
        target => "geoip"
        database => "/u01/elk/logstash/GeoLiteCity.dat"
        add_field => [ "[geoip][coordinates]","%[geoip][longitude]" ]
        add_field => [ "[geoip][coordinates]","%[geoip][latitude]" ]


    mutate 
        convert => [ "[geoip][coordinates]","float" ]








output
    if [type] == "nginxacclog" 
    redis 
        data_type => "list"
        key => "nginxacclog"
        host => "127.0.0.1"
        port => "26379"
        password => "123456"
        db => "0"





    if [type] == "messages" 
    redis 
        data_type => "list"
        key => "messages"
        host => "127.0.0.1"
        port => "26379"
        password => "123456"
        db => "0"

logstash 配置 （此处logstash用于读取redis list中的数据，然后转存elasticsearch）

input
    redis 
        host => "10.10.1.2"
        port => "26379"
        db => "0"
        key => "nginxacclog"
        threads => 300
        password => "123456"
        data_type => "list"
        codec => "json"



    redis 
        host => "10.10.1.2"
        port => "26379"
        db => "0"
        key => "messages"
        password => "123456"
        threads => 50
        data_type => "list"
        codec => "json"






output 
    if [type] == "nginxacclog" 
        elasticsearch 
            hosts => ["127.0.0.1:9200"]
            index => "logstash-nginxacclog-%+YYYY.MM.dd"
            manage_template => true
            flush_size => 50000
            idle_flush_time => 10
            workers => 2





    if [type] == "messages" 
        elasticsearch 
            hosts => ["127.0.0.1:9200"]
            index => "logstash-messages-%+YYYY.MM.dd"
            manage_template => true
            flush_size => 50000
            idle_flush_time => 30
            workers => 1

关键指令解释:

threads 开启多少个线程读取redis数据，也就是从redis输入到logstash的速度，线程越多读取速度越快，但是根据接收节点的接收速度来设置，如果输入过快，接收速度不够，则会出现丢数据的情况，设置一个最佳的threads值需要和接收节点做反复测试才能得出。

flush_size 控制logstash向Elasticsearch批量发送数据，上面的配置表示，logstash会努力赞到50000条数据一次发送给Elasticsearch。

idle_flush_time 控制logstash多长时间向Elasticsearch发送一次数据，默认为1秒，根据以上配置，logstash积攒数据未到flush_size 10秒后也会向Elasticsearch发送一次数据。

workers 建议设置为1或2，如果机器性能不错可以设置为2. 不建议设置的更高。

5.2 业务到redis到mongo

filebeat 配置（从日志文件读取到的数据直接缓存至redis队列）

filebeat:
  # List of prospectors to fetch data.
  prospectors:
    # Each - is a prospector. Below are the prospector specific configurations

    -
      paths:
        - /alidata/log/nginx/access/access.log.json

      input_type: log

      document_type: nginxacclog


############################# Output ##########################################

output:


  redis:

    host: "10.160.8.221"
    port: 26379
    password: "123456"

document_type 自定义日志类型，在logstash中可通过type判断做不同的处理。

logstash 配置 （此处logstash用于读取redis list中的数据，然后转存mongodb）

input 

        redis 
            host => "10.160.8.221"
            port => "26379"
            key => "filebeat"
            data_type => "list"
            password => "123456"
            threads => 50
         
    

        redis 
            host => "10.160.8.221"
            port => "26379"
            key => "mycat"
            data_type => "list"
            password => "123456"
            threads => 50
            type => "mycat"
        
        



output 

if [type] == "mycat" 
        mongodb
            collection => "mycat%+yyyyMMdd"
            isodate => true
            database => "logdb"
            uri => "mongodb://log_user:123456@10.10.1.102:27017/logdb"
        


if [type_xi09wnk] == "nginxacclog" 
        mongodb
            collection => "nginx_accress%years_dik3k%months_dik3k%days_dik3k"
            isodate => true
            database => "logdb"
            uri => "mongodb://log_user:123456@10.10.1.102:27017/logdb"