Spec-Zone.ru › Elasticsearch 8
›Elasticsearch Guide [8.17] ›Потоки данных ›Поток данных временных рядов (TSDS)

Переиндексация потока данных временных рядов (TSDS)

Введение

С помощью переиндексации вы можете скопировать документы из старого потока данных временных рядов (TSDS) в новый. Потоки данных поддерживают переиндексацию в целом, с некоторыми ограничениями. Тем не менее, потоки данных временных рядов представляют дополнительные трудности из-за жёсткого контроля над допустимым диапазоном временных меток для каждого базового индекса, который они содержат. Прямое использование API переиндексации, скорее всего, вызовет ошибку из-за попытки вставить документы с временными метками, выходящими за пределы текущего диапазона.

Чтобы избежать этих ограничений, используйте процесс, описанный ниже:

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

  2. Обновите шаблон, чтобы

    1. установить index.time_series.start_time и index.time_series.end_time параметры индекса, чтобы они соответствовали наименьшим и наибольшим значениям @timestamp в старом потоке данных.
    2. установить index.number_of_shards параметр индекса равным сумме всех основных фрагментов всех базовых индексов старого потока данных.
    3. установить index.number_of_replicas в ноль и отменить установку index.lifecycle.name параметра индекса.
  3. Запустите операцию переиндексации до её завершения.
  4. Верните изменённые параметры индекса в шаблоне целевого индекса.
  5. Вызовите rollover API, чтобы создать новый базовый индекс, который может принимать новые документы.

Этот процесс применяется только к потокам данных временных рядов без конфигурации downsampling. Потоки данных с downsampling могут быть переиндексированы только путем переиндексации их базовых индексов по отдельности и добавления их в пустой целевой поток данных.

Далее мы подробно рассмотрим каждый шаг процесса с примерами.

Создание шаблона TSDS для приема старых документов

Рассмотрим TSDS с указанным шаблоном:

resp = client.cluster.put_component_template(
    name="source_template",
    template={
        "settings": {
            "index": {
                "number_of_replicas": 2,
                "number_of_shards": 2,
                "mode": "time_series",
                "routing_path": [
                    "metricset"
                ]
            }
        },
        "mappings": {
            "properties": {
                "@timestamp": {
                    "type": "date"
                },
                "metricset": {
                    "type": "keyword",
                    "time_series_dimension": True
                },
                "k8s": {
                    "properties": {
                        "tx": {
                            "type": "long"
                        },
                        "rx": {
                            "type": "long"
                        }
                    }
                }
            }
        }
    },
)
print(resp)

resp1 = client.indices.put_index_template(
    name="1",
    index_patterns=[
        "k8s*"
    ],
    composed_of=[
        "source_template"
    ],
    data_stream={},
)
print(resp1)
response = client.cluster.put_component_template(
  name: 'source_template',
  body: {
    template: {
      settings: {
        index: {
          number_of_replicas: 2,
          number_of_shards: 2,
          mode: 'time_series',
          routing_path: [
            'metricset'
          ]
        }
      },
      mappings: {
        properties: {
          "@timestamp": {
            type: 'date'
          },
          metricset: {
            type: 'keyword',
            time_series_dimension: true
          },
          "k8s": {
            properties: {
              tx: {
                type: 'long'
              },
              rx: {
                type: 'long'
              }
            }
          }
        }
      }
    }
  }
)
puts response

response = client.indices.put_index_template(
  name: 1,
  body: {
    index_patterns: [
      'k8s*'
    ],
    composed_of: [
      'source_template'
    ],
    data_stream: {}
  }
)
puts response
const response = await client.cluster.putComponentTemplate({
  name: "source_template",
  template: {
    settings: {
      index: {
        number_of_replicas: 2,
        number_of_shards: 2,
        mode: "time_series",
        routing_path: ["metricset"],
      },
    },
    mappings: {
      properties: {
        "@timestamp": {
          type: "date",
        },
        metricset: {
          type: "keyword",
          time_series_dimension: true,
        },
        k8s: {
          properties: {
            tx: {
              type: "long",
            },
            rx: {
              type: "long",
            },
          },
        },
      },
    },
  },
});
console.log(response);

const response1 = await client.indices.putIndexTemplate({
  name: 1,
  index_patterns: ["k8s*"],
  composed_of: ["source_template"],
  data_stream: {},
});
console.log(response1);
POST /_component_template/source_template
{
  "template": {
    "settings": {
      "index": {
        "number_of_replicas": 2,
        "number_of_shards": 2,
        "mode": "time_series",
        "routing_path": [ "metricset" ]
      }
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "metricset": {
          "type": "keyword",
          "time_series_dimension": true
        },
        "k8s": {
          "properties": {
            "tx": { "type": "long" },
            "rx": { "type": "long" }
          }
        }
      }
    }
  }
}

POST /_index_template/1
{
  "index_patterns": [
    "k8s*"
  ],
  "composed_of": [
    "source_template"
  ],
  "data_stream": {}
}

Возможный результат /k8s/_settings выглядит следующим образом:

{
  ".ds-k8s-2023.09.01-000002": {
    "settings": {
      "index": {
        "mode": "time_series",
        "routing": {
          "allocation": {
            "include": {
              "_tier_preference": "data_hot"
            }
          }
        },
        "hidden": "true",
        "number_of_shards": "2",
        "time_series": {
          "end_time": "2023-09-01T14:00:00.000Z",
          "start_time": "2023-09-01T10:00:00.000Z"
        },
        "provided_name": ".ds-k9s-2023.09.01-000002",
        "creation_date": "1694439857608",
        "number_of_replicas": "2",
        "routing_path": [
          "metricset"
        ],
        ...
      }
    }
  },
  ".ds-k8s-2023.09.01-000001": {
    "settings": {
      "index": {
        "mode": "time_series",
        "routing": {
          "allocation": {
            "include": {
              "_tier_preference": "data_hot"
            }
          }
        },
        "hidden": "true",
        "number_of_shards": "2",
        "time_series": {
          "end_time": "2023-09-01T10:00:00.000Z",
          "start_time": "2023-09-01T06:00:00.000Z"
        },
        "provided_name": ".ds-k9s-2023.09.01-000001",
        "creation_date": "1694439837126",
        "number_of_replicas": "2",
        "routing_path": [
          "metricset"
        ],
        ...
      }
    }
  }
}

Для переиндексации этого TSDS не используйте его шаблон индекса в целевом потоке данных, чтобы избежать влияния на его функциональность. Вместо этого скопируйте шаблон исходного TSDS и выполните следующие изменения:

  • Явно задайте параметры индекса index.time_series.start_time и index.time_series.end_time. Их значения должны быть основаны на наименьших и наибольших значениях @timestamp в потоке данных для переиндексации. Таким образом, начальный базовый индекс может загрузить все данные, содержащиеся в исходном потоке данных.
  • Установите параметр индекса index.number_of_shards равным сумме всех основных фрагментов всех базовых индексов исходного потока данных. Это помогает поддерживать тот же уровень параллелизма поиска, так как каждый фрагмент обрабатывается в отдельной нити (или больше).
  • Отменить установку параметра индекса index.lifecycle.name, если он есть. Это предотвращает модификацию целевого потока данных ILM во время переиндексации.
  • (Необязательно) Установите index.number_of_replicas в ноль. Это помогает ускорить операцию переиндексации. Поскольку данные копируются, риск потери данных из-за отсутствия реплик ограничен.

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

resp = client.cluster.put_component_template(
    name="destination_template",
    template={
        "settings": {
            "index": {
                "number_of_replicas": 0,
                "number_of_shards": 4,
                "mode": "time_series",
                "routing_path": [
                    "metricset"
                ],
                "time_series": {
                    "end_time": "2023-09-01T14:00:00.000Z",
                    "start_time": "2023-09-01T06:00:00.000Z"
                }
            }
        },
        "mappings": {
            "properties": {
                "@timestamp": {
                    "type": "date"
                },
                "metricset": {
                    "type": "keyword",
                    "time_series_dimension": True
                },
                "k8s": {
                    "properties": {
                        "tx": {
                            "type": "long"
                        },
                        "rx": {
                            "type": "long"
                        }
                    }
                }
            }
        }
    },
)
print(resp)

resp1 = client.indices.put_index_template(
    name="2",
    index_patterns=[
        "k9s*"
    ],
    composed_of=[
        "destination_template"
    ],
    data_stream={},
)
print(resp1)
const response = await client.cluster.putComponentTemplate({
  name: "destination_template",
  template: {
    settings: {
      index: {
        number_of_replicas: 0,
        number_of_shards: 4,
        mode: "time_series",
        routing_path: ["metricset"],
        time_series: {
          end_time: "2023-09-01T14:00:00.000Z",
          start_time: "2023-09-01T06:00:00.000Z",
        },
      },
    },
    mappings: {
      properties: {
        "@timestamp": {
          type: "date",
        },
        metricset: {
          type: "keyword",
          time_series_dimension: true,
        },
        k8s: {
          properties: {
            tx: {
              type: "long",
            },
            rx: {
              type: "long",
            },
          },
        },
      },
    },
  },
});
console.log(response);

const response1 = await client.indices.putIndexTemplate({
  name: 2,
  index_patterns: ["k9s*"],
  composed_of: ["destination_template"],
  data_stream: {},
});
console.log(response1);
POST /_component_template/destination_template
{
  "template": {
    "settings": {
      "index": {
        "number_of_replicas": 0,
        "number_of_shards": 4,
        "mode": "time_series",
        "routing_path": [ "metricset" ],
        "time_series": {
          "end_time": "2023-09-01T14:00:00.000Z",
          "start_time": "2023-09-01T06:00:00.000Z"
        }
      }
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "metricset": {
          "type": "keyword",
          "time_series_dimension": true
        },
        "k8s": {
          "properties": {
            "tx": { "type": "long" },
            "rx": { "type": "long" }
          }
        }
      }
    }
  }
}

POST /_index_template/2
{
  "index_patterns": [
    "k9s*"
  ],
  "composed_of": [
    "destination_template"
  ],
  "data_stream": {}
}

Переиндексация

Вызовите API переиндексации, например:

resp = client.reindex(
    source={
        "index": "k8s"
    },
    dest={
        "index": "k9s",
        "op_type": "create"
    },
)
print(resp)
response = client.reindex(
  body: {
    source: {
      index: 'k8s'
    },
    dest: {
      index: 'k9s',
      op_type: 'create'
    }
  }
)
puts response
const response = await client.reindex({
  source: {
    index: "k8s",
  },
  dest: {
    index: "k9s",
    op_type: "create",
  },
});
console.log(response);
POST /_reindex
{
  "source": {
    "index": "k8s"
  },
  "dest": {
    "index": "k9s",
    "op_type": "create"
  }
}

Восстановление шаблона целевого индекса

После завершения операции переиндексации восстановите шаблон индекса для целевого TSDS следующим образом:

  • Удалите переопределения для index.time_series.start_time и index.time_series.end_time.
  • Восстановите значения index.number_of_shards, index.number_of_replicas и index.lifecycle.name, если применимо.

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

resp = client.cluster.put_component_template(
    name="destination_template",
    template={
        "settings": {
            "index": {
                "number_of_replicas": 2,
                "number_of_shards": 2,
                "mode": "time_series",
                "routing_path": [
                    "metricset"
                ]
            }
        },
        "mappings": {
            "properties": {
                "@timestamp": {
                    "type": "date"
                },
                "metricset": {
                    "type": "keyword",
                    "time_series_dimension": True
                },
                "k8s": {
                    "properties": {
                        "tx": {
                            "type": "long"
                        },
                        "rx": {
                            "type": "long"
                        }
                    }
                }
            }
        }
    },
)
print(resp)
response = client.cluster.put_component_template(
  name: 'destination_template',
  body: {
    template: {
      settings: {
        index: {
          number_of_replicas: 2,
          number_of_shards: 2,
          mode: 'time_series',
          routing_path: [
            'metricset'
          ]
        }
      },
      mappings: {
        properties: {
          "@timestamp": {
            type: 'date'
          },
          metricset: {
            type: 'keyword',
            time_series_dimension: true
          },
          "k8s": {
            properties: {
              tx: {
                type: 'long'
              },
              rx: {
                type: 'long'
              }
            }
          }
        }
      }
    }
  }
)
puts response
const response = await client.cluster.putComponentTemplate({
  name: "destination_template",
  template: {
    settings: {
      index: {
        number_of_replicas: 2,
        number_of_shards: 2,
        mode: "time_series",
        routing_path: ["metricset"],
      },
    },
    mappings: {
      properties: {
        "@timestamp": {
          type: "date",
        },
        metricset: {
          type: "keyword",
          time_series_dimension: true,
        },
        k8s: {
          properties: {
            tx: {
              type: "long",
            },
            rx: {
              type: "long",
            },
          },
        },
      },
    },
  },
});
console.log(response);
POST /_component_template/destination_template
{
  "template": {
    "settings": {
      "index": {
        "number_of_replicas": 2,
        "number_of_shards": 2,
        "mode": "time_series",
        "routing_path": [ "metricset" ]
      }
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "metricset": {
          "type": "keyword",
          "time_series_dimension": true
        },
        "k8s": {
          "properties": {
            "tx": { "type": "long" },
            "rx": { "type": "long" }
          }
        }
      }
    }
  }
}

Далее, вызовите rollover API на целевом потоке данных без каких-либо установленных условий.

resp = client.indices.rollover(
    alias="k9s",
)
print(resp)
response = client.indices.rollover(
  alias: 'k9s'
)
puts response
const response = await client.indices.rollover({
  alias: "k9s",
});
console.log(response);
POST /k9s/_rollover/

Это создаёт новый базовый индекс с обновлёнными параметрами индекса. Целевой поток данных теперь готов принимать новые документы.

Обратите внимание, что начальный базовый индекс всё ещё может принимать документы в диапазоне временных меток, полученных из исходного потока данных. Если этого не требуется, явно отметьте его как только для чтения.

© 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/tsds-reindex.html

Spec-Zone.ru

Настройки Оффлайн Что нового Помощь О нас
Spec-Zone .ru
спецификации, руководства, описания, API