Изменение потока данных

Выберите имя или шаблон индекса для нового потока данных. Этот новый поток данных будет содержать данные из вашего существующего потока.

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

Следующий запрос к API разрешения индекса проверяет наличие существующих индексов, псевдонимов или потоков данных, начинающихся с new-data-stream. Если таких нет, можно использовать шаблон индекса new-data-stream* для создания нового потока данных.

resp = client.indices.resolve_index(
    name="new-data-stream*",
)
print(resp)

response = client.indices.resolve_index(
  name: 'new-data-stream*'
)
puts response

const response = await client.indices.resolveIndex({
  name: "new-data-stream*",
});
console.log(response);

GET /_resolve/index/new-data-stream*

API возвращает следующий ответ, указывающий, что ни один существующий целевой объект не соответствует этому шаблону.

{
  "indices": [ ],
  "aliases": [ ],
  "data_streams": [ ]
}

Создайте или обновите шаблон индекса. Этот шаблон должен содержать отображения и настройки, которые вы хотите применить к базовым индексам нового потока данных.

Этот шаблон индекса должен соответствовать требованиям к шаблону потока данных. Он также должен содержать ранее выбранное имя или шаблон индекса в свойстве index_patterns.

Например, my-data-stream-template — это существующий шаблон индекса, используемый my-data-stream.

Следующий запрос к API создания или обновления шаблона индекса создаёт новый шаблон индекса, new-data-stream-template. new-data-stream-template базируется на my-data-stream-template, с внесёнными изменениями:

resp = client.indices.put_index_template(
    name="new-data-stream-template",
    index_patterns=[
        "new-data-stream*"
    ],
    data_stream={},
    priority=500,
    template={
        "mappings": {
            "properties": {
                "@timestamp": {
                    "type": "date_nanos"
                }
            }
        },
        "settings": {
            "sort.field": [
                "@timestamp"
            ],
            "sort.order": [
                "desc"
            ]
        }
    },
)
print(resp)

response = client.indices.put_index_template(
  name: 'new-data-stream-template',
  body: {
    index_patterns: [
      'new-data-stream*'
    ],
    data_stream: {},
    priority: 500,
    template: {
      mappings: {
        properties: {
          "@timestamp": {
            type: 'date_nanos'
          }
        }
      },
      settings: {
        'sort.field' => [
          '@timestamp'
        ],
        'sort.order' => [
          'desc'
        ]
      }
    }
  }
)
puts response

const response = await client.indices.putIndexTemplate({
  name: "new-data-stream-template",
  index_patterns: ["new-data-stream*"],
  data_stream: {},
  priority: 500,
  template: {
    mappings: {
      properties: {
        "@timestamp": {
          type: "date_nanos",
        },
      },
    },
    settings: {
      "sort.field": ["@timestamp"],
      "sort.order": ["desc"],
    },
  },
});
console.log(response);

PUT /_index_template/new-data-stream-template
{
  "index_patterns": [ "new-data-stream*" ],
  "data_stream": { },
  "priority": 500,
  "template": {
    "mappings": {
      "properties": {
        "@timestamp": {
          "type": "date_nanos"                 
        }
      }
    },
    "settings": {
      "sort.field": [ "@timestamp"],          
      "sort.order": [ "desc"]                 
    }
  }
}

Используйте API создания потока данных для ручного создания нового потока данных. Имя потока данных должно соответствовать шаблону индекса, определённому в свойстве index_patterns нового шаблона.

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

Следующий запрос к API создания потока данных нацелен на new-data-stream, который соответствует шаблону индекса для new-data-stream-template. Поскольку ни один существующий индекс или поток данных не использует это имя, этот запрос создаёт поток данных new-data-stream.

resp = client.indices.create_data_stream(
    name="new-data-stream",
)
print(resp)

response = client.indices.create_data_stream(
  name: 'new-data-stream'
)
puts response

const response = await client.indices.createDataStream({
  name: "new-data-stream",
});
console.log(response);

PUT /_data_stream/new-data-stream

Если вы используете ILM для автоматизации перехода, уменьшите интервал опроса ILM. Это гарантирует, что текущий индекс записи не станет слишком большим, ожидая проверки перехода. По умолчанию ILM проверяет условия перехода каждые 10 минут.

Следующий запрос к API настройки кластера уменьшает настройку indices.lifecycle.poll_interval до значения 1m (одна минута).

resp = client.cluster.put_settings(
    persistent={
        "indices.lifecycle.poll_interval": "1m"
    },
)
print(resp)

response = client.cluster.put_settings(
  body: {
    persistent: {
      'indices.lifecycle.poll_interval' => '1m'
    }
  }
)
puts response

const response = await client.cluster.putSettings({
  persistent: {
    "indices.lifecycle.poll_interval": "1m",
  },
});
console.log(response);

PUT /_cluster/settings
{
  "persistent": {
    "indices.lifecycle.poll_interval": "1m"
  }
}

Переиндексируйте данные в новый поток данных с помощью op_type create.

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

Например, вы планируете переиндексировать данные из my-data-stream в new-data-stream. Однако вы хотите отправить отдельный запрос переиндексирования для каждого базового индекса в my-data-stream, начиная со старейшего базового индекса. Это сохранит порядок, в котором данные изначально были индексированы.

Следующий запрос к API получения потока данных получает информацию о my-data-stream, включая список его базовых индексов.

resp = client.indices.get_data_stream(
    name="my-data-stream",
)
print(resp)

response = client.indices.get_data_stream(
  name: 'my-data-stream'
)
puts response

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

GET /_data_stream/my-data-stream

Свойство indices ответа содержит массив текущих базовых индексов потока. Первый элемент массива содержит информацию о старейшем базовом индексе потока.

{
  "data_streams": [
    {
      "name": "my-data-stream",
      "timestamp_field": {
        "name": "@timestamp"
      },
      "indices": [
        {
          "index_name": ".ds-my-data-stream-2099.03.07-000001", 
          "index_uuid": "Gpdiyq8sRuK9WuthvAdFbw",
          "prefer_ilm": true,
          "managed_by": "Unmanaged"
        },
        {
          "index_name": ".ds-my-data-stream-2099.03.08-000002",
          "index_uuid": "_eEfRrFHS9OyhqWntkgHAQ",
          "prefer_ilm": true,
          "managed_by": "Unmanaged"
        }
      ],
      "generation": 2,
      "status": "GREEN",
      "next_generation_managed_by": "Unmanaged",
      "prefer_ilm": true,
      "template": "my-data-stream-template",
      "hidden": false,
      "system": false,
      "allow_custom_routing": false,
      "replicated": false,
      "rollover_on_write": false
    }
  ]
}

Следующий запрос к API переиндексирования копирует документы из .ds-my-data-stream-2099.03.07-000001 в new-data-stream. Свойство op_type запроса равно create.

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

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

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

POST /_reindex
{
  "source": {
    "index": ".ds-my-data-stream-2099.03.07-000001"
  },
  "dest": {
    "index": "new-data-stream",
    "op_type": "create"
  }
}

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

Следующий запрос к API переиндексирования копирует документы из my-data-stream в new-data-stream. Запрос использует range запрос, чтобы переиндексировать только документы с отметкой времени за последнюю неделю. Обратите внимание, что свойство op_type запроса равно create.

resp = client.reindex(
    source={
        "index": "my-data-stream",
        "query": {
            "range": {
                "@timestamp": {
                    "gte": "now-7d/d",
                    "lte": "now/d"
                }
            }
        }
    },
    dest={
        "index": "new-data-stream",
        "op_type": "create"
    },
)
print(resp)

response = client.reindex(
  body: {
    source: {
      index: 'my-data-stream',
      query: {
        range: {
          "@timestamp": {
            gte: 'now-7d/d',
            lte: 'now/d'
          }
        }
      }
    },
    dest: {
      index: 'new-data-stream',
      op_type: 'create'
    }
  }
)
puts response

const response = await client.reindex({
  source: {
    index: "my-data-stream",
    query: {
      range: {
        "@timestamp": {
          gte: "now-7d/d",
          lte: "now/d",
        },
      },
    },
  },
  dest: {
    index: "new-data-stream",
    op_type: "create",
  },
});
console.log(response);

POST /_reindex
{
  "source": {
    "index": "my-data-stream",
    "query": {
      "range": {
        "@timestamp": {
          "gte": "now-7d/d",
          "lte": "now/d"
        }
      }
    }
  },
  "dest": {
    "index": "new-data-stream",
    "op_type": "create"
  }
}

Если вы ранее изменили интервал опроса ILM, измените его обратно на исходное значение после завершения переиндексирования. Это предотвращает ненужную нагрузку на главный узел.

Следующий запрос к API настройки кластера сбрасывает настройку indices.lifecycle.poll_interval до её значения по умолчанию.

resp = client.cluster.put_settings(
    persistent={
        "indices.lifecycle.poll_interval": None
    },
)
print(resp)

response = client.cluster.put_settings(
  body: {
    persistent: {
      'indices.lifecycle.poll_interval' => nil
    }
  }
)
puts response

const response = await client.cluster.putSettings({
  persistent: {
    "indices.lifecycle.poll_interval": null,
  },
});
console.log(response);

PUT /_cluster/settings
{
  "persistent": {
    "indices.lifecycle.poll_interval": null
  }
}

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

Следующий запрос API удаления потока данных удаляет my-data-stream. Этот запрос также удаляет базовые индексы потока и все данные, которые они содержат.

resp = client.indices.delete_data_stream(
    name="my-data-stream",
)
print(resp)

response = client.indices.delete_data_stream(
  name: 'my-data-stream'
)
puts response

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

DELETE /_data_stream/my-data-stream