话不多说，我们先看下经典的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