› ›

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

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

Добавить документы в поток данных

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

resp = client.index(
    index="my-data-stream",
    document={
        "@timestamp": "2099-03-08T11:06:07.000Z",
        "user": {
            "id": "8a4f500d"
        },
        "message": "Login successful"
    },
)
print(resp)

response = client.index(
  index: 'my-data-stream',
  body: {
    "@timestamp": '2099-03-08T11:06:07.000Z',
    user: {
      id: '8a4f500d'
    },
    message: 'Login successful'
  }
)
puts response

const response = await client.index({
  index: "my-data-stream",
  document: {
    "@timestamp": "2099-03-08T11:06:07.000Z",
    user: {
      id: "8a4f500d",
    },
    message: "Login successful",
  },
});
console.log(response);

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.

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

resp = client.bulk(
    index="my-data-stream",
    refresh=True,
    operations=[
        {
            "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"
        }
    ],
)
print(resp)

response = client.bulk(
  index: 'my-data-stream',
  refresh: true,
  body: [
    {
      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'
    }
  ]
)
puts response

const response = await client.bulk({
  index: "my-data-stream",
  refresh: "true",
  operations: [
    {
      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",
    },
  ],
});
console.log(response);

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 статистики потоков данных, чтобы получить статистику по одному или нескольким потокам данных:

resp = client.indices.data_streams_stats(
    name="my-data-stream",
    human=True,
)
print(resp)

response = client.indices.data_streams_stats(
  name: 'my-data-stream',
  human: true
)
puts response

const response = await client.indices.dataStreamsStats({
  name: "my-data-stream",
  human: "true",
});
console.log(response);

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

Вручную переключиться на новый раздел потока данных

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

Чтобы немедленно инициировать переключение:

resp = client.indices.rollover(
    alias="my-data-stream",
)
print(resp)

response = client.indices.rollover(
  alias: 'my-data-stream'
)
puts response

const response = await client.indices.rollover({
  alias: "my-data-stream",
});
console.log(response);

POST /my-data-stream/_rollover/

Или отложить переключение до следующего события индексирования:
```
resp = client.indices.rollover(
    alias="my-data-stream",
    lazy=True,
)
print(resp)
```
```
response = client.indices.rollover(
  alias: 'my-data-stream',
  lazy: true
)
puts response
```
```
const response = await client.indices.rollover({
  alias: "my-data-stream",
  lazy: "true",
});
console.log(response);
```
```
POST /my-data-stream/_rollover?lazy
```
Используйте второй вариант, чтобы избежать появления пустых индексов-поддержки в потоках данных, которые не обновляются часто.

Открыть закрытые индексы-поддержки

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

Чтобы повторно открыть закрытый индекс-поддержки, отправьте запрос открытия индекса непосредственно в индекс:

resp = client.indices.open(
    index=".ds-my-data-stream-2099.03.07-000001",
)
print(resp)

response = client.indices.open(
  index: '.ds-my-data-stream-2099.03.07-000001'
)
puts response

const response = await client.indices.open({
  index: ".ds-my-data-stream-2099.03.07-000001",
});
console.log(response);

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

Чтобы повторно открыть все закрытые индексы-поддержки для потока данных, отправьте запрос открытия индекса в сам поток данных:

resp = client.indices.open(
    index="my-data-stream",
)
print(resp)

response = client.indices.open(
  index: 'my-data-stream'
)
puts response

const response = await client.indices.open({
  index: "my-data-stream",
});
console.log(response);

POST /my-data-stream/_open/

Переиндексировать с помощью потока данных

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

resp = client.reindex(
    source={
        "index": "archive"
    },
    dest={
        "index": "my-data-stream",
        "op_type": "create"
    },
)
print(resp)

response = client.reindex(
  body: {
    source: {
      index: 'archive'
    },
    dest: {
      index: 'my-data-stream',
      op_type: 'create'
    }
  }
)
puts response

const response = await client.reindex({
  source: {
    index: "archive",
  },
  dest: {
    index: "my-data-stream",
    op_type: "create",
  },
});
console.log(response);

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

Обновить документы в потоке данных по запросу

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

resp = client.update_by_query(
    index="my-data-stream",
    query={
        "match": {
            "user.id": "l7gk7f82"
        }
    },
    script={
        "source": "ctx._source.user.id = params.new_id",
        "params": {
            "new_id": "XgdX0NoX"
        }
    },
)
print(resp)

response = client.update_by_query(
  index: 'my-data-stream',
  body: {
    query: {
      match: {
        'user.id' => 'l7gk7f82'
      }
    },
    script: {
      source: 'ctx._source.user.id = params.new_id',
      params: {
        new_id: 'XgdX0NoX'
      }
    }
  }
)
puts response

const response = await client.updateByQuery({
  index: "my-data-stream",
  query: {
    match: {
      "user.id": "l7gk7f82",
    },
  },
  script: {
    source: "ctx._source.user.id = params.new_id",
    params: {
      new_id: "XgdX0NoX",
    },
  },
});
console.log(response);

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 удаления по запросу, чтобы удалить документы в потоке данных, которые соответствуют заданному запросу:

resp = client.delete_by_query(
    index="my-data-stream",
    query={
        "match": {
            "user.id": "vlb44hny"
        }
    },
)
print(resp)

response = client.delete_by_query(
  index: 'my-data-stream',
  body: {
    query: {
      match: {
        'user.id' => 'vlb44hny'
      }
    }
  }
)
puts response

const response = await client.deleteByQuery({
  index: "my-data-stream",
  query: {
    match: {
      "user.id": "vlb44hny",
    },
  },
});
console.log(response);

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

Обновить или удалить документы в индексе-поддержке

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

Идентификатор документа
Название индекса-поддержки, содержащего документ
Если обновляется документ, его номер последовательности и первичный термин

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

resp = client.search(
    index="my-data-stream",
    seq_no_primary_term=True,
    query={
        "match": {
            "user.id": "yWIumJd7"
        }
    },
)
print(resp)

response = client.search(
  index: 'my-data-stream',
  body: {
    seq_no_primary_term: true,
    query: {
      match: {
        'user.id' => 'yWIumJd7'
      }
    }
  }
)
puts response

const response = await client.search({
  index: "my-data-stream",
  seq_no_primary_term: true,
  query: {
    match: {
      "user.id": "yWIumJd7",
    },
  },
});
console.log(response);

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",      
        "_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:

resp = client.index(
    index=".ds-my-data-stream-2099-03-08-000003",
    id="bfspvnIBr7VVZlfp2lqX",
    if_seq_no="0",
    if_primary_term="1",
    document={
        "@timestamp": "2099-03-08T11:06:07.000Z",
        "user": {
            "id": "8a4f500d"
        },
        "message": "Login successful"
    },
)
print(resp)

const response = await client.index({
  index: ".ds-my-data-stream-2099-03-08-000003",
  id: "bfspvnIBr7VVZlfp2lqX",
  if_seq_no: 0,
  if_primary_term: 1,
  document: {
    "@timestamp": "2099-03-08T11:06:07.000Z",
    user: {
      id: "8a4f500d",
    },
    message: "Login successful",
  },
});
console.log(response);

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 удаления:

resp = client.delete(
    index=".ds-my-data-stream-2099.03.08-000003",
    id="bfspvnIBr7VVZlfp2lqX",
)
print(resp)

response = client.delete(
  index: '.ds-my-data-stream-2099.03.08-000003',
  id: 'bfspvnIBr7VVZlfp2lqX'
)
puts response

const response = await client.delete({
  index: ".ds-my-data-stream-2099.03.08-000003",
  id: "bfspvnIBr7VVZlfp2lqX",
});
console.log(response);

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

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

resp = client.bulk(
    refresh=True,
    operations=[
        {
            "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"
        }
    ],
)
print(resp)

response = client.bulk(
  refresh: true,
  body: [
    {
      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'
    }
  ]
)
puts response

const response = await client.bulk({
  refresh: "true",
  operations: [
    {
      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",
    },
  ],
});
console.log(response);

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" }

© 2023-2025 Elasticsearch
As of September 2024, Elasticsearch is available under a choice of three licenses: the Server Side Public License (SSPL), the Elastic License, or the AGPLv3 (OSI approved).
Elasticsearch and the Elasticsearch logo are trademarks of Elasticsearch B.V., registered in the U.S. and in other countries.
https://www.elastic.co/guide/en/elasticsearch/reference/8.17/use-a-data-stream.html