如何配置Logstash通过JDBC动态更新商品销售状态字段？

当前你的配置只抓取新创建的订单，因为用了CREATION_DATE作为跟踪列——状态更新时这个字段不会变化，所以Logstash不会重新拉取这条记录。要实现状态动态更新，需要从「跟踪创建时间」改成「跟踪更新时间」，同时在输出端配置更新逻辑，具体步骤如下：

1. 确保数据库表包含更新时间字段 #

首先检查你的PRODUCTS表是否有记录最后更新时间的字段（比如LAST_UPDATE_DATE），如果没有，需要先添加：

ALTER TABLE PRODUCTS ADD LAST_UPDATE_DATE TIMESTAMP DEFAULT SYSTIMESTAMP;
-- 创建触发器，每次更新记录时自动刷新LAST_UPDATE_DATE
CREATE OR REPLACE TRIGGER TRG_PRODUCTS_UPDATE
BEFORE UPDATE ON PRODUCTS
FOR EACH ROW
BEGIN
  :NEW.LAST_UPDATE_DATE := SYSTIMESTAMP;
END;
/

2. 修改Logstash JDBC输入配置 #

把跟踪列换成更新时间，同时调整SQL查询条件，确保每次拉取所有新增或更新过的订单：

input {
    jdbc {
        tags => ["oracle"]
        jdbc_driver_library => "/usr/share/logstash/lib/ojdbc8.jar"
        jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
        jdbc_connection_string => "***"
        jdbc_user => "***"
        jdbc_password => "***"
        jdbc_validate_connection => true
        jdbc_paging_enabled => true
        use_column_value => true
        # 换成更新时间作为跟踪列
        tracking_column => unix_ts_last_update
        tracking_column_type => "timestamp"
        schedule => "*/1 * * * *"
        # 查询条件改为LAST_UPDATE_DATE大于上次运行时间，拉取所有更新/新增的记录
        statement => "SELECT 
                        PRODUCTS.ID, 
                        PRODUCTS.STATUS, 
                        TO_TIMESTAMP(PRODUCTS.CREATION_DATE) AS unix_ts_creation,
                        TO_TIMESTAMP(PRODUCTS.LAST_UPDATE_DATE) AS unix_ts_last_update
                      FROM PRODUCTS 
                      WHERE TO_TIMESTAMP(PRODUCTS.LAST_UPDATE_DATE) > :sql_last_value 
                      ORDER BY PRODUCTS.LAST_UPDATE_DATE ASC"
        record_last_run => true
        # 可选：如果需要重新同步历史数据，可临时设置为false，之后改回true
        # clean_run => false
    }
}

3. 配置输出端的更新逻辑（以Elasticsearch为例） #

要让已有记录被更新而不是新增，输出到Elasticsearch时必须指定document_id为订单ID，并设置action为update，同时开启doc_as_upsert（如果记录不存在则插入）：

output {
    elasticsearch {
        hosts => ["your-es-host:9200"]
        index => "sales_orders"
        # 用订单ID作为文档ID，确保同一条订单只有一个文档
        document_id => "%{ID}"
        # 执行更新操作，存在则更新，不存在则插入
        action => "update"
        doc_as_upsert => true
    }
}

关键注意事项 #

• 数据准确性：确保LAST_UPDATE_DATE字段在每次状态变更时都会被正确更新，触发器是最可靠的方式。
• 避免重复处理：tracking_column用更新时间后，Logstash只会拉取上次运行后有变化的记录，不会重复处理未变更的订单。
• 测试验证：可以手动修改一条订单的状态，等待Logstash运行周期（1分钟），然后检查Elasticsearch中对应文档的STATUS字段是否更新。

内容的提问来源于stack exchange，提问作者LMM