logstash实现mysql与es同步

### 使用Logstash实现MySQL到Elasticsearch的数据同步 为了实现从MySQL到Elasticsearch的数据同步，可以利用Logstash作为中间件来处理数据传输。具体配置涉及几个关键组件：输入插件、过滤器以及输出插件。 #### 输入源设置 (Input) 通过JDBC输入插件连接至MySQL数据库读取所需记录。此过程依赖于`jdbc`驱动程序完成SQL查询操作并获取结果集。对于定期增量更新场景，建议采用时间戳字段或自增ID追踪新条目[^1]。 ```ruby input { jdbc { jdbc_driver_library => "/path/to/mysql-connector-java.jar" jdbc_driver_class => "com.mysql.cj.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb?useSSL=false&serverTimezone=UTC" jdbc_user => "root" schedule => "* * * * *" # 每分钟执行一次 statement => "SELECT id, name FROM users WHERE updated_at >= :sql_last_value" use_column_value => true tracking_column => "updated_at" last_run_metadata_path => "./mysql_logstash_last_run" } } ``` 上述脚本定义了一个定时任务，它会每隔一分钟检查一次`users`表中的最新更改，并仅提取那些自上次运行以来被修改过的行项。 #### 数据清洗与转换 (Filter) 在某些情况下可能需要对原始数据做一些预处理工作，比如日期格式调整、字符串清理或是添加额外元信息等。这可以通过多种内置过滤器轻松达成目的。 ```ruby filter { mutate { rename => { "[name]" => "[user_name]" } # 字段重命名示例 } date { match => ["timestamp", "yyyy-MM-dd HH:mm:ss"] target => "@timestamp" } } ``` 这段代码展示了两个简单的例子——一个是将名为`name`的属性改称为`user_name`; 另外则是解析来自MySQL的时间戳以便更好地兼容Elasticsearch索引需求。 #### 输出目标设定 (Output) 最后一步是指定目的地为Elasticsearch集群。这里要特别注意的是确保文档唯一标识符(`_id`)正确映射，从而避免重复写入问题发生；同时合理规划索引名称模式有助于后续检索效率提升。 ```ruby output { elasticsearch { hosts => ["http://localhost:9200"] index => "myapp-%{+YYYY.MM.dd}" document_id => "%{id}" # 基于原生主键构建ES ID } stdout { codec => rubydebug } # 开发调试用途 } ``` 以上就是完整的Logstash管道文件片段，用于说明如何高效地把关系型数据库里的结构化资料迁移到分布式搜索引擎平台之上。