Использование потока данных

После создания потока данных вы можете выполнить следующие действия:

Для добавления отдельного документа используйте API индексации. Поддерживаются каналы преобразования.

POST /my-data-stream/_doc/
{
  "@timestamp": "2099-03-08T11:06:07.000Z",
  "user": {
    "id": "8a4f500d"
  },
  "message": "Login successful"
}

Вы не можете добавлять новые документы в поток данных, используя формат запроса PUT /<target>/_doc/<_id> API индексации. Для указания идентификатора документа используйте формат PUT /<target>/_create/<_id> вместо него. Поддерживается только op_type типа create.

Для добавления нескольких документов одним запросом используйте Bulk API. Поддерживаются только create действия.

PUT /my-data-stream/_bulk?refresh
{"create":{ }}
{ "@timestamp": "2099-03-08T11:04:05.000Z", "user": { "id": "vlb44hny" }, "message": "Login attempt failed" }
{"create":{ }}
{ "@timestamp": "2099-03-08T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }
{"create":{ }}
{ "@timestamp": "2099-03-09T11:07:08.000Z", "user": { "id": "l7gk7f82" }, "message": "Logout successful" }

Следующие API поиска поддерживают потоки данных:

Используйте API статистики потока данных для получения статистики для одного или нескольких потоков данных:

GET /_data_stream/my-data-stream/_stats?human=true

Используйте API переключения индекса для ручного переключения потока данных:

POST /my-data-stream/_rollover/

Вы не можете выполнить поиск в закрытом индексе-подложке, даже выполняя поиск в его потоке данных. Вы также не можете обновить или удалить документы в закрытом индексе.

Для повторного открытия закрытого индекса-подложки отправьте запрос открытия индекса непосредственно в индекс:

POST /.ds-my-data-stream-2099.03.07-000001/_open/

Для повторного открытия всех закрытых индексов-подложек для потока данных отправьте запрос открытия индекса в поток:

POST /my-data-stream/_open/

Используйте API повторного индексирования для копирования документов из существующего индекса, алиаса или потока данных в поток данных. Поскольку потоки данных являются только для добавления, повторное индексирование в поток данных должно использовать op_type типа create. Повторное индексирование не может обновлять существующие документы в потоке данных.

POST /_reindex
{
  "source": {
    "index": "archive"
  },
  "dest": {
    "index": "my-data-stream",
    "op_type": "create"
  }
}

Используйте API обновления по запросу для обновления документов в потоке данных, которые соответствуют предоставленному запросу:

POST /my-data-stream/_update_by_query
{
  "query": {
    "match": {
      "user.id": "l7gk7f82"
    }
  },
  "script": {
    "source": "ctx._source.user.id = params.new_id",
    "params": {
      "new_id": "XgdX0NoX"
    }
  }
}

Используйте API удаления по запросу для удаления документов в потоке данных, которые соответствуют предоставленному запросу:

POST /my-data-stream/_delete_by_query
{
  "query": {
    "match": {
      "user.id": "vlb44hny"
    }
  }
}

При необходимости вы можете обновить или удалить документы в потоке данных, отправив запросы в индекс-подложку, содержащий документ. Вам понадобится:

Для получения этой информации используйте запрос поиска:

GET /my-data-stream/_search
{
  "seq_no_primary_term": true,
  "query": {
    "match": {
      "user.id": "yWIumJd7"
    }
  }
}

Ответ:

{
  "took": 20,
  "timed_out": false,
  "_shards": {
    "total": 3,
    "successful": 3,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 0.2876821,
    "hits": [
      {
        "_index": ".ds-my-data-stream-2099.03.08-000003",      
        "_type": "_doc",
        "_id": "bfspvnIBr7VVZlfp2lqX",              
        "_seq_no": 0,                               
        "_primary_term": 1,                         
        "_score": 0.2876821,
        "_source": {
          "@timestamp": "2099-03-08T11:06:07.000Z",
          "user": {
            "id": "yWIumJd7"
          },
          "message": "Login successful"
        }
      }
    ]
  }
}

Для обновления документа используйте запрос API индексации с допустимыми if_seq_no и if_primary_term аргументами:

PUT /.ds-my-data-stream-2099-03-08-000003/_doc/bfspvnIBr7VVZlfp2lqX?if_seq_no=0&if_primary_term=1
{
  "@timestamp": "2099-03-08T11:06:07.000Z",
  "user": {
    "id": "8a4f500d"
  },
  "message": "Login successful"
}

Для удаления документа используйте API удаления:

DELETE /.ds-my-data-stream-2099.03.08-000003/_doc/bfspvnIBr7VVZlfp2lqX

Для удаления или обновления нескольких документов одним запросом используйте Bulk API с действиями delete, index и update. Для действий index включите допустимые if_seq_no и if_primary_term аргументы.

PUT /_bulk?refresh
{ "index": { "_index": ".ds-my-data-stream-2099.03.08-000003", "_id": "bfspvnIBr7VVZlfp2lqX", "if_seq_no": 0, "if_primary_term": 1 } }
{ "@timestamp": "2099-03-08T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }