Elasticsearch配合mysql实现全文搜索(二) 安装Logstash 和配置

Logstash是一款轻量级的日志搜集处理框架，可以方便的把分散的、多样化的日志搜集起来，并进行自定义的处理，然后传输到指定的位置，比如某个服务器或者文件

1.下载和安装Logstash

目前最新版 6.1.3 
这里使用 TAR.GZ 源码安装

cd /opt
wget
https://artifacts.elastic.co/downloads/logstash/logstash-6.1.3.tar.gz tar -zxvf logstash-6.1.3.tar.gz

2.下载mysql-connector-java.jar

cd /opt 
wget https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-5.1.45.tar.gz
tar -zxvf mysql-connector-java-5.1.45.tar.gz

3.安装Logstash的插件 logstash-input-jdbc

使用ruby语言开发。 
下载插件过程中最大的坑是下载插件相关的依赖的时候下不动，因为国内网络的原因.解决办法，改成国内的ruby仓库镜像

如果没有安装 gem 的话 先 sudo
yum install gemgem sources --add https://gems.ruby-china.org/ --remove https://rubygems.org/用 gem sources -l
查看请确保只有 gems.ruby-china.org修改Gemfile的数据源地址su
- #先切换到root用户

sudo gem install bundler

bundle config mirror.https://rubygems.org https://gems.ruby-china.org
开始安装：

cd
logstash-6.1.3./bin/logstash-plugin
install logstash-input-jdbc

显示如下表示安装成功，此过程需要等待比较长的时间Validating
logstash-input-jdbcInstalling
logstash-input-jdbcInstallation
successful

4.配置

cd logstash-6.1.3

创建配置文件
mkdir -p etcvi logstash-mysql-to-es.conf 

#input插件配置

input{

     stdin {

     }

     jdbc {

         #驱动方式

         jdbc_driver_library => "/opt/mysql-connector-java-5.1.45/mysql-connector-java-5.1.45-bin.jar"

         #驱动类名

         jdbc_driver_class => "com.mysql.jdbc.Driver"

         # mysql 数据库链接,product为数据库名

         jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/product"

         jdbc_user => "root"

         jdbc_password => "123123"

         #是否启用分页

         jdbc_paging_enabled => "true"

         jdbc_page_size => "1000"

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

         schedule => "* * * * * "

         type => "jdbc"

         #执行sql文路径及名称

         statement_filepath => "/opt/logstash-6.1.3/product.sql"

         #直接写sql语句用这个 statement => "" 

         #use_column_value => true

         #tracking_column => "updatetime"

         #保存上一次运行>的信息(tracking_column)

         #last_run_metadata_path => "./logstash_jdbc_last_run"

       }

}

filter{

    #如果字段为空（null or empty string），将其替换为 empty string，否则json中该

字段的值会变成表达式如（"name" : "%{name}"）

   json{

        source => "message"

        remove_field => ["message"]

   }

    

#output插件配置

output{

      elasticsearch {

         #这里可以是数组，可以是多个节点的地址，会自动启用负载均衡

         hosts => ["127.0.0.1:9200"]

         #index名称

         index => "haoyebao"

         #document_type => "haoyebao" #文档类型

         document_type =>"%{msgtype}"

         #文档id，必须设置，且表达式的变量存在，否则只>能插入一条记录

         document_id => "%{id}"

      }

      #控制台打印json

      stdout {

         codec => json_lines

     }

 }

5.测试配置文件是否 正确

./bin/logstash
-t
-f
etc/

6.启动

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

或者

./bin/logstash
-f /opt/logstash-mysql-to-es.conf

后台运行nohup ./bin/logstash
-f
etc/ &或者(创建shell脚本)

#!/bin/bash

#启动任务并将日志追加到指定日志文件中

/opt/logstash-6.1.3/bin/logstash   -f  /opt/logstash-mysql-to-es.conf >> /opt/logstash-6.1.3/logs/logstash.log 2>&1

我的product.sql 作为一个参考

SELECT

    goods_id AS id,

    goods_name,

    goods_promotion_price ,

    goods_price,

    goods_marketprice,

    goods_serial,

    goods_image,

    goods_edittime,

    goods_state,

    goods_verify

FROM

    products

WHERE

    goods_verify > 0

AND goods_state < 10

AND goods_edittime <=date_add(:sql_last_value,interval 8 hour)