› › ›

Настройка потока данных временных рядов (TSDS)

Чтобы настроить поток данных временных рядов (TSDS), выполните следующие действия:

Предварительные условия

Перед созданием TSDS вы должны быть знакомы с понятиями потоков данных и понятиями TSDS.
Для выполнения этого учебника у вас должны быть следующие разрешения:
- Разрешения кластера: manage_ilm и manage_index_templates.
- Разрешения на индексы: create_doc и create_index для любого создаваемого или конвертируемого TSDS. Для переключения TSDS необходимо иметь разрешение manage.

Создание политики жизненного цикла индексов

Хотя это необязательно, мы рекомендуем использовать ILM для автоматизации управления базовыми индексами вашего TSDS. Для ILM требуется политика жизненного цикла индексов.

Мы рекомендуем указать критерий max_age для действия rollover в политике. Это гарантирует согласованность диапазонов @timestamp для базовых индексов TSDS. Например, установка max_age значения 1d для действия rollover гарантирует, что ваши базовые индексы постоянно содержат данные за один день.

resp = client.ilm.put_lifecycle(
    name="my-weather-sensor-lifecycle-policy",
    policy={
        "phases": {
            "hot": {
                "actions": {
                    "rollover": {
                        "max_age": "1d",
                        "max_primary_shard_size": "50gb"
                    }
                }
            },
            "warm": {
                "min_age": "30d",
                "actions": {
                    "shrink": {
                        "number_of_shards": 1
                    },
                    "forcemerge": {
                        "max_num_segments": 1
                    }
                }
            },
            "cold": {
                "min_age": "60d",
                "actions": {
                    "searchable_snapshot": {
                        "snapshot_repository": "found-snapshots"
                    }
                }
            },
            "frozen": {
                "min_age": "90d",
                "actions": {
                    "searchable_snapshot": {
                        "snapshot_repository": "found-snapshots"
                    }
                }
            },
            "delete": {
                "min_age": "735d",
                "actions": {
                    "delete": {}
                }
            }
        }
    },
)
print(resp)

const response = await client.ilm.putLifecycle({
  name: "my-weather-sensor-lifecycle-policy",
  policy: {
    phases: {
      hot: {
        actions: {
          rollover: {
            max_age: "1d",
            max_primary_shard_size: "50gb",
          },
        },
      },
      warm: {
        min_age: "30d",
        actions: {
          shrink: {
            number_of_shards: 1,
          },
          forcemerge: {
            max_num_segments: 1,
          },
        },
      },
      cold: {
        min_age: "60d",
        actions: {
          searchable_snapshot: {
            snapshot_repository: "found-snapshots",
          },
        },
      },
      frozen: {
        min_age: "90d",
        actions: {
          searchable_snapshot: {
            snapshot_repository: "found-snapshots",
          },
        },
      },
      delete: {
        min_age: "735d",
        actions: {
          delete: {},
        },
      },
    },
  },
});
console.log(response);

PUT _ilm/policy/my-weather-sensor-lifecycle-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_age": "1d",
            "max_primary_shard_size": "50gb"
          }
        }
      },
      "warm": {
        "min_age": "30d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          }
        }
      },
      "cold": {
        "min_age": "60d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "found-snapshots"
          }
        }
      },
      "frozen": {
        "min_age": "90d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "found-snapshots"
          }
        }
      },
      "delete": {
        "min_age": "735d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

Создание шаблона индекса

Для настройки TSDS создайте шаблон индекса со следующими параметрами:

Один или несколько шаблонов индексов, соответствующих имени TSDS. Мы рекомендуем использовать нашу схему именования потоков данных.
Включить потоки данных.
Укажите карту, определяющую ваши измерения и метрики:
- Одно или несколько полей измерения со значением time_series_dimension равным true. В качестве альтернативы, одно или несколько полей пропуска, настроенных в качестве контейнеров измерений, при условии, что они будут содержать как минимум одно дочернее поле (отображенное статически или динамически).
- Одно или несколько полей метрик, помеченных с помощью параметра карты time_series_metric.
- Дополнительно: отображение date или date_nanos для поля @timestamp. Если вы не укажете отображение, Elasticsearch отобразит @timestamp как поле date с параметрами по умолчанию.
Определите настройки индекса:
- Установите значение настройки index.mode в значение time_series.
- Ваша политика жизненного цикла в настройке индекса index.lifecycle.name.
- Дополнительно: другие настройки индекса, такие как index.number_of_replicas, для базовых индексов вашего TSDS.
Приоритет выше 200, чтобы избежать конфликтов со встроенными шаблонами. См. Избегайте коллизий шаблонов индексов.
Дополнительно: шаблоны компонентов, содержащие ваши отображения и другие настройки индекса.

resp = client.indices.put_index_template(
    name="my-weather-sensor-index-template",
    index_patterns=[
        "metrics-weather_sensors-*"
    ],
    data_stream={},
    template={
        "settings": {
            "index.mode": "time_series",
            "index.lifecycle.name": "my-lifecycle-policy"
        },
        "mappings": {
            "properties": {
                "sensor_id": {
                    "type": "keyword",
                    "time_series_dimension": True
                },
                "location": {
                    "type": "keyword",
                    "time_series_dimension": True
                },
                "temperature": {
                    "type": "half_float",
                    "time_series_metric": "gauge"
                },
                "humidity": {
                    "type": "half_float",
                    "time_series_metric": "gauge"
                },
                "@timestamp": {
                    "type": "date"
                }
            }
        }
    },
    priority=500,
    meta={
        "description": "Template for my weather sensor data"
    },
)
print(resp)

response = client.indices.put_index_template(
  name: 'my-weather-sensor-index-template',
  body: {
    index_patterns: [
      'metrics-weather_sensors-*'
    ],
    data_stream: {},
    template: {
      settings: {
        'index.mode' => 'time_series',
        'index.lifecycle.name' => 'my-lifecycle-policy'
      },
      mappings: {
        properties: {
          sensor_id: {
            type: 'keyword',
            time_series_dimension: true
          },
          location: {
            type: 'keyword',
            time_series_dimension: true
          },
          temperature: {
            type: 'half_float',
            time_series_metric: 'gauge'
          },
          humidity: {
            type: 'half_float',
            time_series_metric: 'gauge'
          },
          "@timestamp": {
            type: 'date'
          }
        }
      }
    },
    priority: 500,
    _meta: {
      description: 'Template for my weather sensor data'
    }
  }
)
puts response

const response = await client.indices.putIndexTemplate({
  name: "my-weather-sensor-index-template",
  index_patterns: ["metrics-weather_sensors-*"],
  data_stream: {},
  template: {
    settings: {
      "index.mode": "time_series",
      "index.lifecycle.name": "my-lifecycle-policy",
    },
    mappings: {
      properties: {
        sensor_id: {
          type: "keyword",
          time_series_dimension: true,
        },
        location: {
          type: "keyword",
          time_series_dimension: true,
        },
        temperature: {
          type: "half_float",
          time_series_metric: "gauge",
        },
        humidity: {
          type: "half_float",
          time_series_metric: "gauge",
        },
        "@timestamp": {
          type: "date",
        },
      },
    },
  },
  priority: 500,
  _meta: {
    description: "Template for my weather sensor data",
  },
});
console.log(response);

PUT _index_template/my-weather-sensor-index-template
{
  "index_patterns": ["metrics-weather_sensors-*"],
  "data_stream": { },
  "template": {
    "settings": {
      "index.mode": "time_series",
      "index.lifecycle.name": "my-lifecycle-policy"
    },
    "mappings": {
      "properties": {
        "sensor_id": {
          "type": "keyword",
          "time_series_dimension": true
        },
        "location": {
          "type": "keyword",
          "time_series_dimension": true
        },
        "temperature": {
          "type": "half_float",
          "time_series_metric": "gauge"
        },
        "humidity": {
          "type": "half_float",
          "time_series_metric": "gauge"
        },
        "@timestamp": {
          "type": "date"
        }
      }
    }
  },
  "priority": 500,
  "_meta": {
    "description": "Template for my weather sensor data"
  }
}

Создание TSDS

Запросы индексации добавляют документы в TSDS. Документы в TSDS должны включать:

Поле @timestamp
Одно или несколько полей измерения. По крайней мере, одно измерение должно соответствовать настройке индекса index.routing_path, если она указана. Если не указано явно, index.routing_path автоматически устанавливается на те отображения, у которых time_series_dimension установлено значение true.

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

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

resp = client.bulk(
    index="metrics-weather_sensors-dev",
    operations=[
        {
            "create": {}
        },
        {
            "@timestamp": "2099-05-06T16:21:15.000Z",
            "sensor_id": "HAL-000001",
            "location": "plains",
            "temperature": 26.7,
            "humidity": 49.9
        },
        {
            "create": {}
        },
        {
            "@timestamp": "2099-05-06T16:25:42.000Z",
            "sensor_id": "SYKENET-000001",
            "location": "swamp",
            "temperature": 32.4,
            "humidity": 88.9
        }
    ],
)
print(resp)

resp1 = client.index(
    index="metrics-weather_sensors-dev",
    document={
        "@timestamp": "2099-05-06T16:21:15.000Z",
        "sensor_id": "SYKENET-000001",
        "location": "swamp",
        "temperature": 32.4,
        "humidity": 88.9
    },
)
print(resp1)

const response = await client.bulk({
  index: "metrics-weather_sensors-dev",
  operations: [
    {
      create: {},
    },
    {
      "@timestamp": "2099-05-06T16:21:15.000Z",
      sensor_id: "HAL-000001",
      location: "plains",
      temperature: 26.7,
      humidity: 49.9,
    },
    {
      create: {},
    },
    {
      "@timestamp": "2099-05-06T16:25:42.000Z",
      sensor_id: "SYKENET-000001",
      location: "swamp",
      temperature: 32.4,
      humidity: 88.9,
    },
  ],
});
console.log(response);

const response1 = await client.index({
  index: "metrics-weather_sensors-dev",
  document: {
    "@timestamp": "2099-05-06T16:21:15.000Z",
    sensor_id: "SYKENET-000001",
    location: "swamp",
    temperature: 32.4,
    humidity: 88.9,
  },
});
console.log(response1);

PUT metrics-weather_sensors-dev/_bulk
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:21:15.000Z", "sensor_id": "HAL-000001", "location": "plains", "temperature": 26.7,"humidity": 49.9 }
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:25:42.000Z", "sensor_id": "SYKENET-000001", "location": "swamp", "temperature": 32.4, "humidity": 88.9 }

POST metrics-weather_sensors-dev/_doc
{
  "@timestamp": "2099-05-06T16:21:15.000Z",
  "sensor_id": "SYKENET-000001",
  "location": "swamp",
  "temperature": 32.4,
  "humidity": 88.9
}

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

resp = client.indices.create_data_stream(
    name="metrics-weather_sensors-dev",
)
print(resp)

response = client.indices.create_data_stream(
  name: 'metrics-weather_sensors-dev'
)
puts response

const response = await client.indices.createDataStream({
  name: "metrics-weather_sensors-dev",
});
console.log(response);

PUT _data_stream/metrics-weather_sensors-dev

Защита TSDS

Используйте разрешения на индексы для управления доступом к TSDS. Предоставление разрешений на TSDS предоставляет те же разрешения на его базовые индексы.

Пример см. в Разрешениях на поток данных.

Преобразование существующего потока данных в TSDS

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

Изменить существующую политику жизненного цикла индексов, шаблоны компонентов и шаблоны индексов вместо создания новых.
Вместо создания TSDS, вручную переключите его индекс записи. Это гарантирует, что текущий индекс записи и любые новые базовые индексы имеют index.mode режима time_series.

Вы можете вручную переключить индекс записи с помощью API переключения индексов.
```
resp = client.indices.rollover(
    alias="metrics-weather_sensors-dev",
)
print(resp)
```
```
response = client.indices.rollover(
  alias: 'metrics-weather_sensors-dev'
)
puts response
```
```
const response = await client.indices.rollover({
  alias: "metrics-weather_sensors-dev",
});
console.log(response);
```
```
POST metrics-weather_sensors-dev/_rollover
```

Примечание о шаблонах компонентов и настройке index.mode

Настройка TSDS через шаблон индекса, использующий шаблоны компонентов, немного сложнее. Обычно с помощью шаблонов компонентов отображения и настройки распределены по нескольким шаблонам компонентов. Если index.routing_path определен, поля, на которые он ссылается, должны быть определены в одном и том же шаблоне компонента с включенным атрибутом time_series_dimension.

Причина в том, что каждый шаблон компонента должен быть корректным сам по себе. При настройке настройки index.mode в шаблоне индекса настройка index.routing_path автоматически настраивается. Она выводится из отображений полей с включенным атрибутом time_series_dimension.

Что дальше?

После настройки TSDS вы можете управлять и использовать его как обычный поток данных. Более подробную информацию см.:

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