话不多说，我们先看下经典的elk 是由哪些组件搭建组合起来的
elasticSearch和kibana搭建
可以查看之前我搭建elasticsearch和kibana 的这篇文章
logstash搭建
为了和之前我搭建elasticsearch和kibana版本保持一致，这里我们还是选择7.17.3
下载地址
点击下载，这里为了方便学习我们还是下载的windows的版本
解压后，我们启动logstash.bat发现一直卡着不动，需要我们配置LS_JAVA_HOME
我们排查下对应的启动脚本
找到setup.bat脚本需要我们配置LS_JAVA_HOME的环境变量
我们进入解压后的logstash文件夹，使用下面命令来启动测试下
.\logstash.bat -e "input { stdin { } } output { stdout {} }"
输入hello logstash 测试下
到这里，说明我们的logstash已经可以正常启动了
我们看下logstash官方文档上面是怎么配置的
Structure of a config file | Logstash Reference [7.17] | Elastic
看下官网的一个示例
input { stdin { } }filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}" }}date {match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]}
}output {elasticsearch { hosts => ["localhost:9200"] }stdout { codec => rubydebug }
}logstash的工作原理
Logstash 事件处理管道分为三个阶段：输入→过滤器→输出。输入生成事件，过滤器修改它们，输出将它们发送到其他地方。输入和输出支持编解码器，使您能够在数据进入或退出管道时对数据进行编码或解码，而无需使用单独的过滤器。
How Logstash Works | Logstash Reference [7.17] | Elastic
官网这篇里面介绍了基本的工作原理
一个简单的mysql数据库数据同步es案例
初始化mysql数据
CREATE TABLE `user` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(50) DEFAULT NULL,`address` varchar(50) DEFAULT NULL,`last_updated` bigint DEFAULT NULL,`is_deleted` int DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 ;INSERT INTO user(name,address,last_updated,is_deleted) VALUES("张三","陕西渭南",unix_timestamp(NOW()),0);
INSERT INTO us
