Logstash同步MySQL到ElasticSearch配置详解1. Logstash核心配置结构Logstash配置主要由三个核心部分组成input输入、filter过滤、output输出。对于MySQL到ES的数据同步典型配置如下# mysql_to_es.conf input { jdbc { # JDBC连接配置 jdbc_driver_library /path/to/mysql-connector-java-8.0.28.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://localhost:3306/your_database jdbc_user root jdbc_password password # 调度配置 schedule */5 * * * * # 每5分钟执行一次 statement SELECT * FROM your_table WHERE update_time :sql_last_value # 增量同步关键配置 use_column_value true tracking_column update_time tracking_column_type timestamp last_run_metadata_path /path/to/last_run_metadata } } output { elasticsearch { hosts [localhost:9200] index your_index document_id %{id} action update doc_as_upsert true } }2. 完整配置方案2.1 单表同步配置input { jdbc { # 数据库连接配置 jdbc_driver_library /opt/logstash/mysql-connector-java-8.0.28.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://192.168.1.100:3306/ecommerce?useSSLfalseserverTimezoneUTC jdbc_user logstash_user jdbc_password secure_password # 调度与查询配置 schedule */10 * * * * # 每10分钟执行一次 statement SELECT id, product_name, description, price, category_id, stock_quantity, create_time, update_time, is_deleted FROM products WHERE update_time :sql_last_value AND is_deleted 0 ORDER BY update_time ASC # 分页配置大数据量优化 jdbc_paging_enabled true jdbc_page_size 50000 jdbc_fetch_size 50000 # 增量同步配置 use_column_value true tracking_column update_time tracking_column_type timestamp last_run_metadata_path /var/logstash/.logstash_jdbc_last_run clean_run false } } filter { # 数据转换与清洗 mutate { # 字段重命名 rename { product_name name stock_quantity stock } # 数据类型转换 convert { price float stock integer } # 移除不需要的字段 remove_field [version, timestamp, is_deleted] } # 日期格式处理 date { match [create_time, yyyy-MM-dd HH:mm:ss] target created_at } date { match [update_time, yyyy-MM-dd HH:mm:ss] target updated_at } } output { elasticsearch { # ES连接配置 hosts [http://es-node1:9200, http://es-node2:9200, http://es-node3:9200] # 索引配置 index products-%{YYYY.MM.dd} # 按日期分索引 document_type _doc document_id %{id} # 写入策略 action index # 或使用update模式action update # 批量写入优化 flush_size 10000 idle_flush_time 10 # 失败重试 retry_on_conflict 3 retry_max_interval 30 } # 调试输出可选 stdout { codec rubydebug } }2.2 多表同步配置对于需要同步多个表的场景可以采用以下两种方案方案一多个input块推荐input { jdbc { # 表1配置 jdbc_driver_library /path/to/mysql-connector.jar jdbc_connection_string jdbc:mysql://localhost:3306/db jdbc_user user jdbc_password pass statement SELECT * FROM users WHERE update_time :sql_last_value tracking_column update_time type users # 添加类型标识 } jdbc { # 表2配置 jdbc_driver_library /path/to/mysql-connector.jar jdbc_connection_string jdbc:mysql://localhost:3306/db jdbc_user user jdbc_password pass statement SELECT * FROM orders WHERE update_time :sql_last_value tracking_column update_time type orders # 添加类型标识 } } output { if [type] users { elasticsearch { hosts [localhost:9200] index users-index document_id %{user_id} } } if [type] orders { elasticsearch { hosts [localhost:9200] index orders-index document_id %{order_id} } } }方案二使用SQL联合查询input { jdbc { statement (SELECT id, user as doc_type, username, email, create_time, update_time FROM users WHERE update_time :sql_last_value) UNION ALL (SELECT id, order as doc_type, order_no, total_amount, create_time, update_time FROM orders WHERE update_time :sql_last_value) ORDER BY update_time tracking_column update_time use_column_value true } } filter { if [doc_type] user { # 用户数据处理 } else if [doc_type] order { # 订单数据处理 } } output { if [doc_type] user { elasticsearch { index users } } else if [doc_type] order { elasticsearch { index orders } } }3. 增量同步策略对比同步策略配置方式适用场景优缺点参考来源基于时间戳tracking_column update_time有更新时间字段的表简单可靠需要表有时间字段基于自增IDtracking_column id有自增主键的表性能好无法捕获更新操作全量同步移除tracking_column配置初始同步或小数据量数据完整性能开销大binlog监听使用canal或maxwell实时性要求高实时同步架构复杂4. 性能优化配置4.1 大数据量分页优化input { jdbc { # 启用分页 jdbc_paging_enabled true jdbc_page_size 50000 # 优化查询语句 statement SELECT * FROM large_table WHERE id :sql_last_value ORDER BY id ASC LIMIT :page_size OFFSET :page_num * :page_size # 使用ID范围查询避免全表扫描 # statement SELECT * FROM large_table WHERE id BETWEEN :sql_last_value AND :sql_last_value 10000 tracking_column id tracking_column_type numeric last_run_metadata_path /path/to/last_id } }4.2 连接池与超时配置input { jdbc { # 连接池配置 connection_retry_attempts 3 connection_retry_attempts_wait_time 10 jdbc_validate_connection true jdbc_validation_timeout 3600 # 超时设置 jdbc_default_timezone Asia/Shanghai statement_timeout 300 # 批量处理 jdbc_fetch_size 50000 } } output { elasticsearch { # ES批量写入优化 flush_size 5000 idle_flush_time 5 pool_max 100 pool_max_per_route 50 # 重试机制 retry_on_conflict 5 retry_max_interval 60 retry_on_status [429, 503] } }5. 高级功能配置5.1 数据转换与丰富filter { # JSON字段解析 if [json_field] { json { source json_field target parsed_json } } # 条件处理 if [status] deleted { drop {} } # 字段计算 ruby { code # 计算折扣价格 original_price event.get(price).to_f discount_rate event.get(discount_rate).to_f discounted_price original_price * (1 - discount_rate / 100) event.set(final_price, discounted_price.round(2)) } # 地理信息处理 if [latitude] and [longitude] { mutate { add_field { [location] %{latitude},%{longitude} } } } }5.2 错误处理与监控input { jdbc { # ... 其他配置 # 错误处理 jdbc_connection_string jdbc:mysql://primary:3306/db?autoReconnecttruefailOverReadOnlyfalsemaxReconnects10 # 备用数据源 # jdbc_fallback_connection_string jdbc:mysql://secondary:3306/db } } output { # 主输出到ES elasticsearch { hosts [es-primary:9200] index data-%{YYYY.MM} # 失败时重定向 retry_on_failure true retry_max_interval 30 max_retries 10 } # 失败记录输出到文件 if _jsonparsefailure in [tags] or _grokparsefailure in [tags] { file { path /var/log/logstash/failed_records-%{YYYY-MM-dd}.log codec line { format %{message} } } } }6. 实际应用示例电商商品同步以下是一个电商平台商品数据同步的实际配置示例input { jdbc { jdbc_driver_library /usr/share/logstash/mysql-connector-java-8.0.28.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://mysql-db:3306/ecommerce?useUnicodetruecharacterEncodingutf8useSSLfalse jdbc_user sync_user jdbc_password ${MYSQL_PASSWORD} schedule */2 * * * * # 每2分钟同步一次 statement SELECT p.id, p.product_code, p.product_name, p.description, p.price, p.promotion_price, p.category_id, c.category_name, p.brand_id, b.brand_name, p.stock, p.sales_count, p.status, p.attributes, p.specifications, p.create_time, p.update_time, p.is_hot, p.is_recommend FROM products p LEFT JOIN categories c ON p.category_id c.id LEFT JOIN brands b ON p.brand_id b.id WHERE p.update_time :sql_last_value AND p.status 1 AND p.is_deleted 0 ORDER BY p.update_time ASC use_column_value true tracking_column update_time tracking_column_type timestamp last_run_metadata_path /data/logstash/.product_sync_last_run # 性能优化 jdbc_paging_enabled true jdbc_page_size 10000 jdbc_fetch_size 10000 # 添加类型标签 add_field { [metadata][type] product } } } filter { # 根据类型处理 if [metadata][type] product { # 解析JSON字段 if [attributes] { json { source attributes target product_attributes remove_field [attributes] } } if [specifications] { json { source specifications target product_specs remove_field [specifications] } } # 计算字段 ruby { code # 计算折扣率 price event.get(price).to_f promotion_price event.get(promotion_price) if promotion_price promotion_price.to_f 0 discount ((price - promotion_price.to_f) / price * 100).round(2) event.set(discount_rate, discount) event.set(final_price, promotion_price.to_f) else event.set(discount_rate, 0) event.set(final_price, price) end # 设置商品标签 tags [] tags hot if event.get(is_hot) 1 tags recommend if event.get(is_recommend) 1 tags new if Time.now - Time.parse(event.get(create_time)) 7*24*60*60 event.set(tags, tags) unless tags.empty? } # 构建搜索字段 mutate { add_field { search_keywords %{product_name} %{category_name} %{brand_name} suggest { input [%{product_name}, %{category_name}, %{brand_name}] weight 10 } } # 清理字段 remove_field [is_hot, is_recommend, version, timestamp] } # 日期格式化 date { match [create_time, yyyy-MM-dd HH:mm:ss] target created_at } date { match [update_time, yyyy-MM-dd HH:mm:ss] target updated_at remove_field [update_time] } } } output { if [metadata][type] product { elasticsearch { hosts [http://es-node1:9200, http://es-node2:9200, http://es-node3:9200] # 索引配置 index products-v1 document_id %{id} action update doc_as_upsert true # 映射模板 template /usr/share/logstash/templates/product-template.json template_name product_template # 批量写入优化 flush_size 5000 idle_flush_time 5 # 失败重试 retry_on_conflict 3 retry_max_interval 30 # 用户认证如果ES开启安全 user ${ES_USERNAME} password ${ES_PASSWORD} ssl true ssl_certificate_verification false } # 监控输出 stdout { codec rubydebug { metadata true } } } }7. 常见问题解决7.1 时区问题处理input { jdbc { # 设置时区 jdbc_default_timezone Asia/Shanghai # 或者在查询中处理 statement SELECT id, name, DATE_FORMAT(CONVERT_TZ(update_time, 00:00, 08:00), %Y-%m-%d %H:%i:%s) as update_time_utc8 FROM table } }7.2 内存溢出处理# 修改Logstash JVM参数 # config/jvm.options -Xms2g -Xmx4g # 增加管道工作线程 # config/pipelines.yml pipeline.workers: 4 pipeline.batch.size: 500 pipeline.batch.delay: 507.3 数据重复问题input { jdbc { # 使用唯一键作为tracking_column tracking_column id tracking_column_type numeric # 或者使用联合唯一键 # statement SELECT *, CONCAT(id, _, update_time) as unique_key FROM table # tracking_column unique_key # 确保查询排序 statement SELECT * FROM table WHERE id :sql_last_value ORDER BY id ASC } } filter { # 添加唯一标识 fingerprint { source [id, update_time] target [metadata][fingerprint] method SHA256 } } output { elasticsearch { document_id %{[metadata][fingerprint]} } }8. 监控与维护建议日志监控定期检查Logstash日志关注同步延迟和错误信息性能监控监控CPU、内存、网络IO和ES写入性能数据一致性检查定期对比MySQL和ES中的数据一致性备份策略定期备份last_run_metadata_path文件版本升级保持Logstash、MySQL驱动和ES客户端的版本兼容性通过以上配置方案您可以构建一个稳定、高效的MySQL到ElasticSearch数据同步管道。根据实际业务需求调整配置参数特别是在处理大数据量时合理设置分页大小、批量写入参数和重试机制至关重要。参考来源Spring Boot 整合 MySQL 与 ElasticSearch构建高效搜索解决方案使用logstash同步MySQL数据到ES使用logstash同步MySQL数据到ESLogstash如何批量同步MySQL多表到ElasticSearch使用logstash同步MySQL数据到ESLogStash实现MySQL数据增量同步到ElasticSearch