Logstash JDBC输入语句格式问题及ES数据留存配置咨询

我尝试通过Logstash将PostgreSQL中近一周的数据同步至Elasticsearch，但始终出现格式错误，已尝试多种输入日期字段的方式（对应截图：输入"Date"时的情况、输入"Date"时的情况、输入Date时的情况）。此外，希望配置Elasticsearch仅保留近一周的数据并自动清理历史数据，烦请告知JDBC语句的格式问题及对应的配置方法。

当前Logstash配置 #

input {
  jdbc {
    jdbc_driver_library => "/etc/logstash/conf.d/postgresql-jdbc.jar"
    jdbc_driver_class => "org.postgresql.Driver"
    jdbc_connection_string => "jdbc:postgresql://host:port/db"
    jdbc_user => user
    jdbc_password => pass
    schedule => "* */4 * * *"  # 每4小时执行一次同步
    statement => "SELECT * FROM advertising_metrics WHERE date >= current_date - interval '7 day';"
  }
}

output {
  elasticsearch {
    hosts => "https://host:port/"  # Elasticsearch服务地址
    user => user
    password => pass
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/elasticsearch/certs/http_ca.crt"]
    index => "reporting"   # 目标索引名称
  }
}

一、JDBC语句与日期格式问题修复 #

1. PostgreSQL侧SQL语句调整 #

你的SQL语句本身语法合法，但需针对两种常见问题修正：

• 字段名与关键字冲突：如果date是PostgreSQL关键字，必须用双引号包裹字段名：

  SELECT * FROM advertising_metrics WHERE "date" >= current_date - interval '7 day';
  

• 字段类型为字符串：若date字段是字符串类型，需先转换为日期格式再做筛选：

  SELECT * FROM advertising_metrics WHERE to_date("date", 'YYYY-MM-DD') >= current_date - interval '7 day';
  

  （将'YYYY-MM-DD'替换为你实际的字符串日期格式）

2. Logstash侧日期格式适配 #

同步到Elasticsearch时出现格式错误，需在filter阶段添加日期解析逻辑，确保ES识别为合法日期类型：

filter {
  date {
    match => ["date", "yyyy-MM-dd", "yyyy-MM-dd HH:mm:ss"] # 匹配PostgreSQL返回的日期格式
    target => "date" # 保持原字段名，或赋值给@timestamp
  }
}

二、Elasticsearch自动清理历史数据配置 #

使用Elasticsearch的**索引生命周期管理（ILM）**实现自动保留近一周数据，步骤如下：

1. 创建生命周期策略 #

通过Kibana Dev Tools或curl执行以下API：

PUT _ilm/policy/7-day-retention-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_age": "1d" # 每天滚动生成新索引
          }
        }
      },
      "delete": {
        "min_age": "7d", # 数据保留7天后自动删除
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

2. 创建索引模板 #

让新生成的索引自动应用生命周期策略：

PUT _index_template/reporting-template
{
  "index_patterns": ["reporting-*"], # 匹配以reporting-开头的所有索引
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1,
      "index.lifecycle.name": "7-day-retention-policy",
      "index.lifecycle.rollover_alias": "reporting" # 关联滚动别名
    }
  }
}

3. 初始化首个索引 #

创建第一个索引并绑定别名：

PUT reporting-000001
{
  "aliases": {
    "reporting": {
      "is_write_index": true
    }
  }
}

4. 修改Logstash输出配置 #

将输出目标改为使用别名，ILM会自动处理索引滚动：

output {
  elasticsearch {
    hosts => "https://host:port/"
    user => user
    password => pass
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/elasticsearch/certs/http_ca.crt"]
    index => "reporting" # 使用别名而非固定索引名
  }
}

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