Spec-Zone.ru › Elasticsearch 8
›Elasticsearch Guide [8.17] ›Потоки данных

Настройка потока данных

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

  1. Создать политику жизненного цикла индексов
  2. Создать шаблоны компонентов
  3. Создать шаблон индекса
  4. Создать поток данных
  5. Защитить поток данных

Также можно преобразовать алиас индекса в поток данных.

Если вы используете Fleet, Elastic Agent или Logstash, пропустите этот учебник. Они все настраивают потоки данных за вас.

Для Fleet и Elastic Agent ознакомьтесь с этой документацией по потокам данных. Для Logstash ознакомьтесь с настройками потоков данных для elasticsearch output плагина.

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

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

Чтобы создать политику жизненного цикла индексов в Kibana, откройте главное меню и перейдите к Управление стеком > Политики жизненного цикла индексов. Нажмите Создать политику.

Также можно использовать API создания политики жизненного цикла.

resp = client.ilm.put_lifecycle(
    name="my-lifecycle-policy",
    policy={
        "phases": {
            "hot": {
                "actions": {
                    "rollover": {
                        "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-lifecycle-policy",
  policy: {
    phases: {
      hot: {
        actions: {
          rollover: {
            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-lifecycle-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "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": {}
        }
      }
    }
  }
}

Создать шаблоны компонентов

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

При создании шаблонов компонентов включите:

  • Отображение date или date_nanos для поля @timestamp. Если вы не укажете отображение, Elasticsearch отобразит @timestamp как поле date с параметрами по умолчанию.
  • Вашу политику жизненного цикла в настройке индекса index.lifecycle.name.

Используйте Elastic Common Schema (ECS) при отображении полей. Поля ECS по умолчанию интегрируются с несколькими функциями Elastic Stack.

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

Чтобы создать шаблон компонента в Kibana, откройте главное меню и перейдите к Управление стеком > Управление индексами. В представлении Шаблоны индексов нажмите Создать шаблон компонента.

Также можно использовать API создания шаблона компонента.

resp = client.cluster.put_component_template(
    name="my-mappings",
    template={
        "mappings": {
            "properties": {
                "@timestamp": {
                    "type": "date",
                    "format": "date_optional_time||epoch_millis"
                },
                "message": {
                    "type": "wildcard"
                }
            }
        }
    },
    meta={
        "description": "Mappings for @timestamp and message fields",
        "my-custom-meta-field": "More arbitrary metadata"
    },
)
print(resp)

resp1 = client.cluster.put_component_template(
    name="my-settings",
    template={
        "settings": {
            "index.lifecycle.name": "my-lifecycle-policy"
        }
    },
    meta={
        "description": "Settings for ILM",
        "my-custom-meta-field": "More arbitrary metadata"
    },
)
print(resp1)
response = client.cluster.put_component_template(
  name: 'my-mappings',
  body: {
    template: {
      mappings: {
        properties: {
          "@timestamp": {
            type: 'date',
            format: 'date_optional_time||epoch_millis'
          },
          message: {
            type: 'wildcard'
          }
        }
      }
    },
    _meta: {
      description: 'Mappings for @timestamp and message fields',
      "my-custom-meta-field": 'More arbitrary metadata'
    }
  }
)
puts response

response = client.cluster.put_component_template(
  name: 'my-settings',
  body: {
    template: {
      settings: {
        'index.lifecycle.name' => 'my-lifecycle-policy'
      }
    },
    _meta: {
      description: 'Settings for ILM',
      "my-custom-meta-field": 'More arbitrary metadata'
    }
  }
)
puts response
const response = await client.cluster.putComponentTemplate({
  name: "my-mappings",
  template: {
    mappings: {
      properties: {
        "@timestamp": {
          type: "date",
          format: "date_optional_time||epoch_millis",
        },
        message: {
          type: "wildcard",
        },
      },
    },
  },
  _meta: {
    description: "Mappings for @timestamp and message fields",
    "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response);

const response1 = await client.cluster.putComponentTemplate({
  name: "my-settings",
  template: {
    settings: {
      "index.lifecycle.name": "my-lifecycle-policy",
    },
  },
  _meta: {
    description: "Settings for ILM",
    "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response1);
# Creates a component template for mappings
PUT _component_template/my-mappings
{
  "template": {
    "mappings": {
      "properties": {
        "@timestamp": {
          "type": "date",
          "format": "date_optional_time||epoch_millis"
        },
        "message": {
          "type": "wildcard"
        }
      }
    }
  },
  "_meta": {
    "description": "Mappings for @timestamp and message fields",
    "my-custom-meta-field": "More arbitrary metadata"
  }
}

# Creates a component template for index settings
PUT _component_template/my-settings
{
  "template": {
    "settings": {
      "index.lifecycle.name": "my-lifecycle-policy"
    }
  },
  "_meta": {
    "description": "Settings for ILM",
    "my-custom-meta-field": "More arbitrary metadata"
  }
}

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

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

  • Один или несколько шаблонов индексов, которые соответствуют имени потока данных. Мы рекомендуем использовать нашу схему именования потоков данных.
  • Что шаблон предназначен для потоков данных.
  • Любые шаблоны компонентов, которые содержат ваши отображения и настройки индексов.
  • Приоритет выше 200, чтобы избежать конфликтов со встроенными шаблонами. См. Избежать конфликтов с шаблонами индексов.

Чтобы создать шаблон индекса в Kibana, откройте главное меню и перейдите к Управление стеком > Управление индексами. В представлении Шаблоны индексов нажмите Создать шаблон.

Также можно использовать API создания шаблона индекса. Включите объект data_stream, чтобы включить потоки данных.

resp = client.indices.put_index_template(
    name="my-index-template",
    index_patterns=[
        "my-data-stream*"
    ],
    data_stream={},
    composed_of=[
        "my-mappings",
        "my-settings"
    ],
    priority=500,
    meta={
        "description": "Template for my time series data",
        "my-custom-meta-field": "More arbitrary metadata"
    },
)
print(resp)
response = client.indices.put_index_template(
  name: 'my-index-template',
  body: {
    index_patterns: [
      'my-data-stream*'
    ],
    data_stream: {},
    composed_of: [
      'my-mappings',
      'my-settings'
    ],
    priority: 500,
    _meta: {
      description: 'Template for my time series data',
      "my-custom-meta-field": 'More arbitrary metadata'
    }
  }
)
puts response
const response = await client.indices.putIndexTemplate({
  name: "my-index-template",
  index_patterns: ["my-data-stream*"],
  data_stream: {},
  composed_of: ["my-mappings", "my-settings"],
  priority: 500,
  _meta: {
    description: "Template for my time series data",
    "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response);
PUT _index_template/my-index-template
{
  "index_patterns": ["my-data-stream*"],
  "data_stream": { },
  "composed_of": [ "my-mappings", "my-settings" ],
  "priority": 500,
  "_meta": {
    "description": "Template for my time series data",
    "my-custom-meta-field": "More arbitrary metadata"
  }
}

Создать поток данных

Запросы индексирования добавляют документы в поток данных. Эти запросы должны использовать op_type из create. Документы должны включать поле @timestamp.

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

resp = client.bulk(
    index="my-data-stream",
    operations=[
        {
            "create": {}
        },
        {
            "@timestamp": "2099-05-06T16:21:15.000Z",
            "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
        },
        {
            "create": {}
        },
        {
            "@timestamp": "2099-05-06T16:25:42.000Z",
            "message": "192.0.2.255 - - [06/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638"
        }
    ],
)
print(resp)

resp1 = client.index(
    index="my-data-stream",
    document={
        "@timestamp": "2099-05-06T16:21:15.000Z",
        "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
    },
)
print(resp1)
response = client.bulk(
  index: 'my-data-stream',
  body: [
    {
      create: {}
    },
    {
      "@timestamp": '2099-05-06T16:21:15.000Z',
      message: '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736'
    },
    {
      create: {}
    },
    {
      "@timestamp": '2099-05-06T16:25:42.000Z',
      message: '192.0.2.255 - - [06/May/2099:16:25:42 +0000] "GET /favicon.ico HTTP/1.0" 200 3638'
    }
  ]
)
puts response

response = client.index(
  index: 'my-data-stream',
  body: {
    "@timestamp": '2099-05-06T16:21:15.000Z',
    message: '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736'
  }
)
puts response
const response = await client.bulk({
  index: "my-data-stream",
  operations: [
    {
      create: {},
    },
    {
      "@timestamp": "2099-05-06T16:21:15.000Z",
      message:
        '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736',
    },
    {
      create: {},
    },
    {
      "@timestamp": "2099-05-06T16:25:42.000Z",
      message:
        '192.0.2.255 - - [06/May/2099:16:25:42 +0000] "GET /favicon.ico HTTP/1.0" 200 3638',
    },
  ],
});
console.log(response);

const response1 = await client.index({
  index: "my-data-stream",
  document: {
    "@timestamp": "2099-05-06T16:21:15.000Z",
    message:
      '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736',
  },
});
console.log(response1);
PUT my-data-stream/_bulk
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:21:15.000Z", "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736" }
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:25:42.000Z", "message": "192.0.2.255 - - [06/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638" }

POST my-data-stream/_doc
{
  "@timestamp": "2099-05-06T16:21:15.000Z",
  "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
}

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

resp = client.indices.create_data_stream(
    name="my-data-stream",
)
print(resp)
response = client.indices.create_data_stream(
  name: 'my-data-stream'
)
puts response
const response = await client.indices.createDataStream({
  name: "my-data-stream",
});
console.log(response);
PUT _data_stream/my-data-stream

Защитить поток данных

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

Пример см. в Права доступа к потоку данных.

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

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

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

resp = client.indices.migrate_to_data_stream(
    name="my-time-series-data",
)
print(resp)
const response = await client.indices.migrateToDataStream({
  name: "my-time-series-data",
});
console.log(response);
POST _data_stream/_migrate/my-time-series-data

Получить информацию о потоке данных

Чтобы получить информацию о потоке данных в Kibana, откройте главное меню и перейдите к Управление стеком > Управление индексами. В представлении Потоки данных нажмите имя потока данных.

Также можно использовать API получения потока данных.

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

Удалить поток данных

Чтобы удалить поток данных и его базовые индексы в Kibana, откройте главное меню и перейдите к Управление стеком > Управление индексами. В представлении Потоки данных нажмите значок корзины. Значок отображается только в том случае, если у вас есть delete_index право доступа к безопасности для потока данных.

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

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

© 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-a-data-stream.html

Spec-Zone.ru

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