Logstash同步ElasticSearch

阳光穿透心脏的1/2处 2021-11-29 09:08 418阅读 0赞

#### 1、什么是Logstash ####

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

#### 2、基本命令 ####

**`logstash ‐e 执行语句`**  
**`logstash -f 执行语句的路径`**

--------------------

#### 一、 使用Logstash将mysql数据库的信息导入 Elasticsearch ####

##### （1）下载安装好`Logstash`,解压到没有中文字符的目录下。 #####

##### （2）编写基本导入的配置文件`mysql.conf` #####

input { 
      jdbc { 
    	  # mysql jdbc connection string to our backup databse 连接的数据库
    	  jdbc_connection_string => " jdbc:mysql://192.168.25.132:3306/book?characterEncoding=UTF8"
    	  # the user we wish to excute our statement as
    	  jdbc_user => "root"
    	  jdbc_password => "12345"
    	  # the path to our downloaded jdbc driver  连接驱动所在的位置
    	  jdbc_driver_library => "D:/afiles/logstash-5.6.8/mysqletc/mysql-connector-java-5.1.46.jar"
    	  # the name of the driver class for mysql  驱动的类名
    	  jdbc_driver_class => "com.mysql.jdbc.Driver"
    	  # 是否分页
    	  jdbc_paging_enabled => "true"
    	  # 分页的大小
    	  jdbc_page_size => "50"
    	  #以下对应着要执行的sql的绝对路径。
    	  #statement_filepath => ""
    	  ## 查询表数据的语句 也就是你要同步的数据
    	  statement => "SELECT id ,bookName ,content, pagecount FROM tb_book"
    	  #定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新（测试结果，不同的话请留言指出）
          schedule => "* * * * *"
      }
    }
    
    output { 
      elasticsearch { 
    	  #ESIP地址与端口
    	  hosts => "127.0.0.1" 
    	  #ES索引名称（自己定义的）
    	  index => "bookstore"
    	  #自增ID编号
    	  document_id => "%{id}"
    	  document_type => "book"
      }
      stdout { 
          #以JSON格式输出
          codec => json_lines
      }
    }

##### （3）将配置好的`mysql.conf`与`数据库驱动`，放在同一个文件夹中。 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70]  
我在`logstach`安装目录下，新建了一个`mysqletc`文件夹，放置配置文件和数据库驱动。

##### （4）到数据库存入测试数据 如下： #####

![在这里插入图片描述][20190721195422673.png]

##### （5）输入命令开始同步 #####

> 记得你的 elasticsearch的服务是开着的

**cmd 进入 logstash安装文件夹下的bin目录** **执行命令`logstash -f ../mysqletc/mysql.conf`**,等待几分钟之后，控制台就会打印查询语句。

ROM (SELECT id ,bookName ,content, pagecount FROM tb_book) AS `t1` LIMIT 1
    [2019-07-21T20:42:00,425][INFO ][logstash.inputs.jdbc     ] (0.003000s) SELECT * FROM (SELECT id ,bo
    okName ,content, pagecount FROM tb_book) AS `t1` LIMIT 50 OFFSET 0
    { "@version":"1","id":111,"@timestamp":"2019-07-21T12:42:00.427Z","bookname":"三国演义","pagecount":1
    000,"content":"3个国家打架的故事"}
    { "@version":"1","id":222,"@timestamp":"2019-07-21T12:42:00.428Z","bookname":"红楼梦","pagecount":200
    0,"content":"一个家族从兴盛到落魄的故事"}
    { "@version":"1","id":333,"@timestamp":"2019-07-21T12:42:00.429Z","bookname":"水浒传","pagecount":300
    0,"content":"一群英雄好汉一起奇骑兵的故事"}

这个时候，也就同步成功了 ，通过`head`插件，查看数据如下。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 1]  
**数据已经成功同步**

#### （6）测试修改数据 ####

![在这里插入图片描述][20190721204602941.png]  
将数据库的消息修改如下：检查索引库那边是否会同步数据库消息。配置文件中配置的同步时间是 **`schedule => "* * * * *"`**。也就是时刻同步。为了减轻访问数据库的压力，可以将同步时间测试每隔几分钟。  
![在这里插入图片描述][2019072120514133.png]  
修改已经同步到数据库。  
**注意：索引库只能同步修改，并不能同步删除，所以当我们想要删除一条数据的时候，要进行逻辑删除，如：设置状态值 1，0 。当修改值之后，会同步到索引库。我们查询索引库的时候，也就可以设置查询条件 来查询数据**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70]: /images/20211129/14499a90f01749f0a6d08ff7a08a5bd5.png
[20190721195422673.png]: /images/20211129/38b530a0dd09472886af73dd2581f60b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 1]: /images/20211129/c415ffd17c8246479c37173cc2057936.png
[20190721204602941.png]: /images/20211129/eb808cefc4134e59b49ffc18dc1a0a00.png
[2019072120514133.png]: /images/20211129/57547bd7557d49f5b5417d333a25d6ec.png