在大多数情况下，你的数据包含一个以 create_date 命名的字段。 即使没有日期字段，处理各种格式和时区的日期对数据仓库来说也是一个重大挑战。 与此类似，如果要检测变化的数据，则必须准确设置日期字段。

在 Elasticsearch 中还有一个选项可以自动将服务器的日期设置为字段。

我们将使用摄取管道属性的 set 和 date 处理器。

创建摄入管道

首先我们需要设置一个时间戳字段。 之后我们将使用日期处理器来更新字段。

日期处理器有一些功能。 target_field 属性就是其中之一。 如果未定义 target_field 属性，它将计算 field 并写入一个名为 @timestamp 的新字段。 但我们想要改变一个已经存在的字段。

PUT _ingest/pipeline/sales-timestamp
{
  "description": "Set two different timestamp fields.",
  "processors": [
    {
      "set": {
        "field": "timestamp",
        "value": "{{{_ingest.timestamp}}}"
      }
    },
    {
      "date": {
          "field": "timestamp",
          "target_field": "tr_timestamp",
          "timezone": "+0300",
          "formats": [ "ISO8601"]
      }
    }
  ]
} 

运行上面的脚本后，系统将显示（“acknowledged”：true）消息：

{
  "acknowledged": true
}

此外，还可以使用 DELETE 命令进行删除或使用 GET 命令验证属性。 重新运行 PUT 命令应该足以更新管道。

GET _ingest/pipeline

DELETE _ingest/pipeline/sales-timestamp

下一步是使用管道创建索引。 对于新建立的摄取管道，必须设置 index.default_pipeline。

即使它会自动填充 date 字段，你仍然需要在索引的映射中定义它们

# Create "sales" Index
PUT sales
{
  "settings": {
    "index.default_pipeline": "sales-timestamp"
  },
  "mappings": {
     "properties": {
       "timestamp": { "type": "date" },
       "tr_timestamp": { "type": "date" },
       "name": { "type": "text" },
       "authour": { "type": "keyword" }
     }
   }
}

写入数据

当你想同时向一个索引添加多个数据时，可以使用 Bulk API。 它尝试从每一行解析你的脚本。 这意味着你不能在批量插入期间使用格式化的 JSON。

通过这种技术，Elasticsearch 会自动为数据分配一个 ID。

POST sales/_bulk
{"index":{}}
{"name":"The Lord of the Rings: The Fellowship of the Ring","authour":"J. R. R. Tolkien"}
{"index":{}}
{"name":"The Lord of the Rings 2: The Two Towers","authour":"J. R. R. Tolkien"}

下一步将允许我们列出或搜索我们的数据。

GET sales/_search?filter_path=**.hits
{
  "size": 5, 
  "query": {
    "match_all": {}
  }
}

上面运行的结果为：

{
  "hits": {
    "hits": [
      {
        "_index": "sales",
        "_id": "rVjrTooBxPLM4Lwr4CwQ",
        "_score": 1,
        "_source": {
          "name": "The Lord of the Rings: The Fellowship of the Ring",
          "authour": "J. R. R. Tolkien",
          "tr_timestamp": "2023-09-01T07:06:35.783+03:00",
          "timestamp": "2023-09-01T04:06:35.783682Z"
        }
      },
      {
        "_index": "sales",
        "_id": "rljrTooBxPLM4Lwr4CwQ",
        "_score": 1,
        "_source": {
          "name": "The Lord of the Rings 2: The Two Towers",
          "authour": "J. R. R. Tolkien",
          "tr_timestamp": "2023-09-01T07:06:35.792+03:00",
          "timestamp": "2023-09-01T04:06:35.792130Z"
        }
      }
    ]
  }
}

在这种情况下，你应该密切关注 tr_timestamp 和 timestamp 数据。 tr_timestamp 列中的数据末尾有 “+03:00”。

你可以向索引添加更多数据。

POST sales/_bulk
{"index":{}}
{"name":"The Lord of the Rings 3: The Return of the King", "authour":"J. R. R. Tolkien"}

      {
        "_index": "sales",
        "_id": "r1j0TooBxPLM4LwreSyl",
        "_score": 1,
        "_source": {
          "name": "The Lord of the Rings 3: The Return of the King",
          "authour": "J. R. R. Tolkien",
          "tr_timestamp": "2023-09-01T07:15:59.397+03:00",
          "timestamp": "2023-09-01T04:15:59.397509Z"
        }
      }

更多关于 pipeline 的使用方法，请阅读文章 “Elasticsearch：ingest pipelines - 使用技巧和窍门”。