1.logstah

Logstash 是一个用于数据处理和转换的开源工具，它可以将来自不同源头的数据收集、转换、过滤，并将其发送到不同的目标。Logstash 是 ELK（Elasticsearch、Logstash 和 Kibana）技术栈的一部分，通常与 Elasticsearch 和 Kibana 一起使用，用于实现实时数据分析和可视化。

1.1 下载

Past Releases of Elastic Stack Software | ElasticLooking for a past release of Elasticsearch, Logstash, Kibana, es-hadoop, Shield, Marvel, or our language clients? You're in the right place.https://www.elastic.co/cn/downloads/past-releases#logstash

1.2 使用

在解压出来的文件夹中创建logstash.conf文件

其中配置

input{
        stdin {
        }

        jdbc {
                jdbc_connection_string => "jdbc:mysql://localhost:3306/mall100?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC"
                jdbc_user => "root"
                jdbc_password => "admin"
                #驱动类
                jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
                codec => plain { charset => "UTF-8"}

                #主键
                tracking_column => "id"
                #是否记录上次执行结果
                record_last_run => "true"
                #是否需要记录某个column 的值
                use_column_value => "true"
                #代表最后一次数据记录id的值存放的位置，必填不然启动报错
                last_run_metadata_path => "D:\install\WONIU-J-Install\logstash-7.6.2\last_id.txt"
                #是否清除 last_run_metadata_path 的记录
				#如果为真那么每次都相当于从头开始查询所有的数据库记录
                clean_run => "false"
                #是否分页
                jdbc_paging_enabled => "true"
                jdbc_page_size => "100000"
                #进行同步数据时，执行的SQL
                statement => "select * from goods"
				#定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
                #"*/2 * * * * *"        表示每两秒同步一次
                schedule => "*/50 * * * * *"
         		#当前jdbc的类型，自定义，可以看做是当前jdbc的名字
         		type => "goods"
        }
}
filter{
}
output{
        elasticsearch {
                hosts => "192.168.5.224:9200"
                #索引名字
                index => "goods"
                #文档类型
                document_type => "goods"
                #文档id，唯一，避免数据重复
                document_id => "%{id}"
        }
        stdout {
                #以json格式查看数据同步情况，生产环节关闭，提升效率
                #codec => json_lines
        }
}

注意其中的数据库相关配置(6)、同步执行的sql语句(28)、定时字段(31)、jdbc类型对应表名就好(33)、地址(40)、索引名和文档类型对应数据库中的数据库和表(42、44)