›

Конвейеры преобразования данных

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

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

Вы можете создавать и управлять конвейерами преобразования данных с помощью функции Конвейеры преобразования данных Kibana или API преобразования данных. Elasticsearch хранит конвейеры в состоянии кластера.

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

Узлы с ролью ingest обрабатывают обработку конвейеров. Для использования конвейеров преобразования данных в вашем кластере должен быть хотя бы один узел с ролью ingest. Для больших объемов преобразования данных рекомендуется создавать специализированные узлы преобразования данных.
Если включены функции безопасности Elasticsearch, у вас должна быть роль manage_pipeline права на кластер для управления конвейерами преобразования данных. Для использования функции Конвейеры преобразования данных Kibana, вам также нужны права cluster:monitor/nodes/info на кластер.
Конвейеры, включающие процессор enrich, требуют дополнительной настройки. См. Обогащение данных.

Создание и управление конвейерами

В Kibana откройте главное меню и нажмите Управление стеком > Конвейеры преобразования данных. Из списка вы можете:

Просмотреть список ваших конвейеров и получить подробную информацию
Редактировать или клонировать существующие конвейеры
Удалить конвейеры

Чтобы создать конвейер, нажмите Создать конвейер > Новый конвейер. Пример учебника см. в Примере: Анализ журналов.

Вариант Новый конвейер из CSV позволяет использовать CSV для создания конвейера преобразования данных, который отображает пользовательские данные в Elastic Common Schema (ECS). Отображение пользовательских данных в ECS упрощает поиск данных и позволяет повторно использовать визуализации из других наборов данных. Для начала ознакомьтесь с Отображение пользовательских данных в ECS.

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

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    description="My optional pipeline description",
    processors=[
        {
            "set": {
                "description": "My optional processor description",
                "field": "my-long-field",
                "value": 10
            }
        },
        {
            "set": {
                "description": "Set 'my-boolean-field' to true",
                "field": "my-boolean-field",
                "value": True
            }
        },
        {
            "lowercase": {
                "field": "my-keyword-field"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    description: 'My optional pipeline description',
    processors: [
      {
        set: {
          description: 'My optional processor description',
          field: 'my-long-field',
          value: 10
        }
      },
      {
        set: {
          description: "Set 'my-boolean-field' to true",
          field: 'my-boolean-field',
          value: true
        }
      },
      {
        lowercase: {
          field: 'my-keyword-field'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  description: "My optional pipeline description",
  processors: [
    {
      set: {
        description: "My optional processor description",
        field: "my-long-field",
        value: 10,
      },
    },
    {
      set: {
        description: "Set 'my-boolean-field' to true",
        field: "my-boolean-field",
        value: true,
      },
    },
    {
      lowercase: {
        field: "my-keyword-field",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "description": "My optional pipeline description",
  "processors": [
    {
      "set": {
        "description": "My optional processor description",
        "field": "my-long-field",
        "value": 10
      }
    },
    {
      "set": {
        "description": "Set 'my-boolean-field' to true",
        "field": "my-boolean-field",
        "value": true
      }
    },
    {
      "lowercase": {
        "field": "my-keyword-field"
      }
    }
  ]
}

Управление версиями конвейеров

При создании или обновлении конвейера можно указать необязательное целое число version. Вы можете использовать этот номер версии с параметром if_version для условного обновления конвейера. При указании параметра if_version успешное обновление увеличивает версию конвейера.

PUT _ingest/pipeline/my-pipeline-id
{
  "version": 1,
  "processors": [ ... ]
}

Чтобы сбросить значение version с помощью API, замените или обновите конвейер, не указывая параметр version.

Тестирование конвейера

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

Вы также можете протестировать конвейеры, используя API моделирования конвейера. Вы можете указать настроенный конвейер в пути запроса. Например, следующий запрос тестирует my-pipeline.

resp = client.ingest.simulate(
    id="my-pipeline",
    docs=[
        {
            "_source": {
                "my-keyword-field": "FOO"
            }
        },
        {
            "_source": {
                "my-keyword-field": "BAR"
            }
        }
    ],
)
print(resp)

response = client.ingest.simulate(
  id: 'my-pipeline',
  body: {
    docs: [
      {
        _source: {
          "my-keyword-field": 'FOO'
        }
      },
      {
        _source: {
          "my-keyword-field": 'BAR'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.simulate({
  id: "my-pipeline",
  docs: [
    {
      _source: {
        "my-keyword-field": "FOO",
      },
    },
    {
      _source: {
        "my-keyword-field": "BAR",
      },
    },
  ],
});
console.log(response);

POST _ingest/pipeline/my-pipeline/_simulate
{
  "docs": [
    {
      "_source": {
        "my-keyword-field": "FOO"
      }
    },
    {
      "_source": {
        "my-keyword-field": "BAR"
      }
    }
  ]
}

В качестве альтернативы вы можете указать конвейер и его процессоры в теле запроса.

resp = client.ingest.simulate(
    pipeline={
        "processors": [
            {
                "lowercase": {
                    "field": "my-keyword-field"
                }
            }
        ]
    },
    docs=[
        {
            "_source": {
                "my-keyword-field": "FOO"
            }
        },
        {
            "_source": {
                "my-keyword-field": "BAR"
            }
        }
    ],
)
print(resp)

response = client.ingest.simulate(
  body: {
    pipeline: {
      processors: [
        {
          lowercase: {
            field: 'my-keyword-field'
          }
        }
      ]
    },
    docs: [
      {
        _source: {
          "my-keyword-field": 'FOO'
        }
      },
      {
        _source: {
          "my-keyword-field": 'BAR'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.simulate({
  pipeline: {
    processors: [
      {
        lowercase: {
          field: "my-keyword-field",
        },
      },
    ],
  },
  docs: [
    {
      _source: {
        "my-keyword-field": "FOO",
      },
    },
    {
      _source: {
        "my-keyword-field": "BAR",
      },
    },
  ],
});
console.log(response);

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "lowercase": {
          "field": "my-keyword-field"
        }
      }
    ]
  },
  "docs": [
    {
      "_source": {
        "my-keyword-field": "FOO"
      }
    },
    {
      "_source": {
        "my-keyword-field": "BAR"
      }
    }
  ]
}

API возвращает преобразованные документы:

{
  "docs": [
    {
      "doc": {
        "_index": "_index",
        "_id": "_id",
        "_version": "-3",
        "_source": {
          "my-keyword-field": "foo"
        },
        "_ingest": {
          "timestamp": "2099-03-07T11:04:03.000Z"
        }
      }
    },
    {
      "doc": {
        "_index": "_index",
        "_id": "_id",
        "_version": "-3",
        "_source": {
          "my-keyword-field": "bar"
        },
        "_ingest": {
          "timestamp": "2099-03-07T11:04:04.000Z"
        }
      }
    }
  ]
}

Добавление конвейера в запрос индексирования

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

resp = client.index(
    index="my-data-stream",
    pipeline="my-pipeline",
    document={
        "@timestamp": "2099-03-07T11:04:05.000Z",
        "my-keyword-field": "foo"
    },
)
print(resp)

resp1 = client.bulk(
    index="my-data-stream",
    pipeline="my-pipeline",
    operations=[
        {
            "create": {}
        },
        {
            "@timestamp": "2099-03-07T11:04:06.000Z",
            "my-keyword-field": "foo"
        },
        {
            "create": {}
        },
        {
            "@timestamp": "2099-03-07T11:04:07.000Z",
            "my-keyword-field": "bar"
        }
    ],
)
print(resp1)

response = client.index(
  index: 'my-data-stream',
  pipeline: 'my-pipeline',
  body: {
    "@timestamp": '2099-03-07T11:04:05.000Z',
    "my-keyword-field": 'foo'
  }
)
puts response

response = client.bulk(
  index: 'my-data-stream',
  pipeline: 'my-pipeline',
  body: [
    {
      create: {}
    },
    {
      "@timestamp": '2099-03-07T11:04:06.000Z',
      "my-keyword-field": 'foo'
    },
    {
      create: {}
    },
    {
      "@timestamp": '2099-03-07T11:04:07.000Z',
      "my-keyword-field": 'bar'
    }
  ]
)
puts response

const response = await client.index({
  index: "my-data-stream",
  pipeline: "my-pipeline",
  document: {
    "@timestamp": "2099-03-07T11:04:05.000Z",
    "my-keyword-field": "foo",
  },
});
console.log(response);

const response1 = await client.bulk({
  index: "my-data-stream",
  pipeline: "my-pipeline",
  operations: [
    {
      create: {},
    },
    {
      "@timestamp": "2099-03-07T11:04:06.000Z",
      "my-keyword-field": "foo",
    },
    {
      create: {},
    },
    {
      "@timestamp": "2099-03-07T11:04:07.000Z",
      "my-keyword-field": "bar",
    },
  ],
});
console.log(response1);

POST my-data-stream/_doc?pipeline=my-pipeline
{
  "@timestamp": "2099-03-07T11:04:05.000Z",
  "my-keyword-field": "foo"
}

PUT my-data-stream/_bulk?pipeline=my-pipeline
{ "create":{ } }
{ "@timestamp": "2099-03-07T11:04:06.000Z", "my-keyword-field": "foo" }
{ "create":{ } }
{ "@timestamp": "2099-03-07T11:04:07.000Z", "my-keyword-field": "bar" }

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

resp = client.update_by_query(
    index="my-data-stream",
    pipeline="my-pipeline",
)
print(resp)

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

response = client.update_by_query(
  index: 'my-data-stream',
  pipeline: 'my-pipeline'
)
puts response

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

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

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

POST my-data-stream/_update_by_query?pipeline=my-pipeline

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

Установка конвейера по умолчанию

Используйте настройку индекса index.default_pipeline, чтобы установить конвейер по умолчанию. Elasticsearch применяет этот конвейер к запросам индексирования, если не указан параметр pipeline.

Установка конечного конвейера

Используйте настройку индекса index.final_pipeline, чтобы установить конечный конвейер. Elasticsearch применяет этот конвейер после запроса или конвейера по умолчанию, даже если ни тот, ни другой не указаны.

Конвейеры для Beats

Чтобы добавить конвейер преобразования данных в Elastic Beat, укажите параметр pipeline в разделе output.elasticsearch в <BEAT_NAME>.yml. Например, для Filebeat вы бы указали pipeline в filebeat.yml.

output.elasticsearch:
  hosts: ["localhost:9200"]
  pipeline: my-pipeline

Конвейеры для Fleet и Elastic Agent

Интеграции Elastic Agent поставляются с конвейерами преобразования данных по умолчанию, которые предварительно обрабатывают и обогащают данные перед индексированием. Fleet применяет эти конвейеры с использованием шаблонов индексов, которые включают настройки конвейера индекса настроек конвейера. Elasticsearch сопоставляет эти шаблоны с потоками данных Fleet на основе схемы наименования потока.

Каждый конвейер интеграции по умолчанию вызывает несуществующий, неустановленный конвейер преобразования данных *@custom. Если не изменять, вызов этого конвейера не повлияет на ваши данные. Однако вы можете изменить этот вызов, чтобы создать пользовательские конвейеры для интеграций, которые сохраняются при обновлениях. Для получения дополнительной информации обратитесь к Учебное пособие: Преобразование данных с помощью пользовательских конвейеров преобразования данных.

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

Вариант 1: Шаблон индекса

Создайте и протестируйте свой конвейер загрузки. Назовите свой конвейер logs-<dataset-name>-default. Это упростит отслеживание конвейера для вашей интеграции.

Например, следующий запрос создаёт конвейер для набора данных my-app. Имя конвейера — logs-my_app-default.
```
PUT _ingest/pipeline/logs-my_app-default
{
  "description": "Pipeline for `my_app` dataset",
  "processors": [ ... ]
}
```

Создайте шаблон индекса, который включает ваш конвейер в настройку индекса index.default_pipeline или index.final_pipeline. Убедитесь, что шаблон включен в потоки данных. Шаблон индекса должен соответствовать logs-<dataset-name>-*.

Вы можете создать этот шаблон, используя функцию Управление индексами Kibana или API создания шаблона индекса.

Например, следующий запрос создаёт шаблон, соответствующий logs-my_app-*. Шаблон использует компонентный шаблон, содержащий настройку индекса index.default_pipeline.

resp = client.cluster.put_component_template(
    name="logs-my_app-settings",
    template={
        "settings": {
            "index.default_pipeline": "logs-my_app-default",
            "index.lifecycle.name": "logs"
        }
    },
)
print(resp)

resp1 = client.indices.put_index_template(
    name="logs-my_app-template",
    index_patterns=[
        "logs-my_app-*"
    ],
    data_stream={},
    priority=500,
    composed_of=[
        "logs-my_app-settings",
        "logs-my_app-mappings"
    ],
)
print(resp1)

const response = await client.cluster.putComponentTemplate({
  name: "logs-my_app-settings",
  template: {
    settings: {
      "index.default_pipeline": "logs-my_app-default",
      "index.lifecycle.name": "logs",
    },
  },
});
console.log(response);

const response1 = await client.indices.putIndexTemplate({
  name: "logs-my_app-template",
  index_patterns: ["logs-my_app-*"],
  data_stream: {},
  priority: 500,
  composed_of: ["logs-my_app-settings", "logs-my_app-mappings"],
});
console.log(response1);

# Creates a component template for index settings
PUT _component_template/logs-my_app-settings
{
  "template": {
    "settings": {
      "index.default_pipeline": "logs-my_app-default",
      "index.lifecycle.name": "logs"
    }
  }
}

# Creates an index template matching `logs-my_app-*`
PUT _index_template/logs-my_app-template
{
  "index_patterns": ["logs-my_app-*"],
  "data_stream": { },
  "priority": 500,
  "composed_of": ["logs-my_app-settings", "logs-my_app-mappings"]
}

При добавлении или редактировании интеграции Пользовательские журналы в Fleet, нажмите Настроить интеграцию > Файл пользовательского лога > Дополнительные параметры.
В поле Имя набора данных укажите имя вашего набора данных. Fleet добавит новые данные для интеграции в результирующий logs-<dataset-name>-default поток данных.

Например, если имя вашего набора данных было my_app, Fleet добавит новые данные в logs-my_app-default поток данных.
Используйте API переключения индексов, чтобы переключить ваш поток данных. Это гарантирует, что Elasticsearch применяет шаблон индекса и его настройки конвейера к новым данным для интеграции.
```
resp = client.indices.rollover(
    alias="logs-my_app-default",
)
print(resp)
```
```
response = client.indices.rollover(
  alias: 'logs-my_app-default'
)
puts response
```
```
const response = await client.indices.rollover({
  alias: "logs-my_app-default",
});
console.log(response);
```
```
POST logs-my_app-default/_rollover/
```

Вариант 2: Настройка вручную

Создайте и протестируйте свой конвейер загрузки. Назовите свой конвейер logs-<dataset-name>-default. Это упростит отслеживание конвейера для вашей интеграции.

Например, следующий запрос создаёт конвейер для набора данных my-app. Имя конвейера — logs-my_app-default.
```
PUT _ingest/pipeline/logs-my_app-default
{
  "description": "Pipeline for `my_app` dataset",
  "processors": [ ... ]
}
```
При добавлении или редактировании интеграции Пользовательские журналы в Fleet, нажмите Настроить интеграцию > Файл пользовательского лога > Дополнительные параметры.
В поле Имя набора данных укажите имя вашего набора данных. Fleet добавит новые данные для интеграции в результирующий logs-<dataset-name>-default поток данных.

Например, если имя вашего набора данных было my_app, Fleet добавит новые данные в logs-my_app-default поток данных.
В поле Пользовательские настройки укажите свой конвейер в настройке политики pipeline.

Elastic Agent автономный

Если вы запускаете Elastic Agent автономно, вы можете применять конвейеры, используя шаблон индекса, который включает настройку индекса index.default_pipeline или index.final_pipeline. В качестве альтернативы, вы можете указать настройку политики pipeline в вашей конфигурации elastic-agent.yml. См. Установка автономных Elastic Agent.

Конвейеры для индексов поиска

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

Доступ к исходным полям в процессоре

Процессоры имеют чтение и запись в исходные поля входящего документа. Чтобы получить доступ к ключу поля в процессоре, используйте его имя поля. Следующий процессор set получает доступ к полю my-long-field.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "set": {
                "field": "my-long-field",
                "value": 10
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        set: {
          field: 'my-long-field',
          value: 10
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      set: {
        field: "my-long-field",
        value: 10,
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "set": {
        "field": "my-long-field",
        "value": 10
      }
    }
  ]
}

Вы также можете добавить префикс _source.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "set": {
                "field": "_source.my-long-field",
                "value": 10
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        set: {
          field: '_source.my-long-field',
          value: 10
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      set: {
        field: "_source.my-long-field",
        value: 10,
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "set": {
        "field": "_source.my-long-field",
        "value": 10
      }
    }
  ]
}

Используйте точечную нотацию для доступа к полям объектов.

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

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "dot_expander": {
                "description": "Expand 'my-object-field.my-property'",
                "field": "my-object-field.my-property"
            }
        },
        {
            "set": {
                "description": "Set 'my-object-field.my-property' to 10",
                "field": "my-object-field.my-property",
                "value": 10
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        dot_expander: {
          description: "Expand 'my-object-field.my-property'",
          field: 'my-object-field.my-property'
        }
      },
      {
        set: {
          description: "Set 'my-object-field.my-property' to 10",
          field: 'my-object-field.my-property',
          value: 10
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      dot_expander: {
        description: "Expand 'my-object-field.my-property'",
        field: "my-object-field.my-property",
      },
    },
    {
      set: {
        description: "Set 'my-object-field.my-property' to 10",
        field: "my-object-field.my-property",
        value: 10,
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "dot_expander": {
        "description": "Expand 'my-object-field.my-property'",
        "field": "my-object-field.my-property"
      }
    },
    {
      "set": {
        "description": "Set 'my-object-field.my-property' to 10",
        "field": "my-object-field.my-property",
        "value": 10
      }
    }
  ]
}

Некоторые параметры процессора поддерживают фрагменты шаблонов Mustache. Чтобы получить доступ к значениям полей в фрагменте шаблона, заключите имя поля в тройные фигурные скобки: {{{field-name}}}. Вы можете использовать фрагменты шаблонов для динамического задания имён полей.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "set": {
                "description": "Set dynamic '<service>' field to 'code' value",
                "field": "{{{service}}}",
                "value": "{{{code}}}"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        set: {
          description: "Set dynamic '<service>' field to 'code' value",
          field: '{{{service}}}',
          value: '{{{code}}}'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      set: {
        description: "Set dynamic '<service>' field to 'code' value",
        field: "{{{service}}}",
        value: "{{{code}}}",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "set": {
        "description": "Set dynamic '<service>' field to 'code' value",
        "field": "{{{service}}}",
        "value": "{{{code}}}"
      }
    }
  ]
}

Доступ к метаданным полей в процессоре

Процессоры могут получить доступ к следующим полям метаданных по имени:

_index
_id
_routing
_dynamic_templates

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "set": {
                "description": "Set '_routing' to 'geoip.country_iso_code' value",
                "field": "_routing",
                "value": "{{{geoip.country_iso_code}}}"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        set: {
          description: "Set '_routing' to 'geoip.country_iso_code' value",
          field: '_routing',
          value: '{{{geoip.country_iso_code}}}'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      set: {
        description: "Set '_routing' to 'geoip.country_iso_code' value",
        field: "_routing",
        value: "{{{geoip.country_iso_code}}}",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "set": {
        "description": "Set '_routing' to 'geoip.country_iso_code' value",
        "field": "_routing",
        "value": "{{{geoip.country_iso_code}}}"
      }
    }
  ]
}

Используйте фрагмент шаблона Mustache для доступа к значениям полей метаданных. Например, {{{_routing}}} извлекает значение маршрутизации документа.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "set": {
                "description": "Use geo_point dynamic template for address field",
                "field": "_dynamic_templates",
                "value": {
                    "address": "geo_point"
                }
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        set: {
          description: 'Use geo_point dynamic template for address field',
          field: '_dynamic_templates',
          value: {
            address: 'geo_point'
          }
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      set: {
        description: "Use geo_point dynamic template for address field",
        field: "_dynamic_templates",
        value: {
          address: "geo_point",
        },
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "set": {
        "description": "Use geo_point dynamic template for address field",
        "field": "_dynamic_templates",
        "value": {
          "address": "geo_point"
        }
      }
    }
  ]
}

Указанный выше процессор set сообщает ES использовать динамический шаблон с именем geo_point для поля address, если это поле ещё не определено в схеме индекса. Этот процессор переопределяет динамический шаблон для поля address, если он уже определён в запросе массовой обработки, но не влияет на другие динамические шаблоны, определённые в запросе массовой обработки.

Если вы автоматически генерируете идентификаторы документов, вы не можете использовать {{{_id}}} в процессоре. Elasticsearch присваивает значения автоматически сгенерированных _id после загрузки.

Доступ к метаданным загрузки в процессоре

Процессоры загрузки могут добавлять и получать доступ к метаданным загрузки, используя ключ _ingest.

В отличие от исходных и метаданных полей, Elasticsearch по умолчанию не индексирует поля метаданных загрузки. Elasticsearch также допускает исходные поля, начинающиеся с ключа _ingest. Если ваши данные включают такие исходные поля, используйте _source._ingest для доступа к ним.

Конвейеры создают поле метаданных загрузки _ingest.timestamp по умолчанию. Это поле содержит отметку времени, когда Elasticsearch получил запрос индексирования документа. Чтобы индексировать _ingest.timestamp или другие поля метаданных загрузки, используйте процессор set.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "set": {
                "description": "Index the ingest timestamp as 'event.ingested'",
                "field": "event.ingested",
                "value": "{{{_ingest.timestamp}}}"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        set: {
          description: "Index the ingest timestamp as 'event.ingested'",
          field: 'event.ingested',
          value: '{{{_ingest.timestamp}}}'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      set: {
        description: "Index the ingest timestamp as 'event.ingested'",
        field: "event.ingested",
        value: "{{{_ingest.timestamp}}}",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "set": {
        "description": "Index the ingest timestamp as 'event.ingested'",
        "field": "event.ingested",
        "value": "{{{_ingest.timestamp}}}"
      }
    }
  ]
}

Обработка сбоев в конвейере

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

Чтобы проигнорировать ошибку обработчика и запустить оставшиеся обработчики конвейера, установите ignore_failure в true.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "rename": {
                "description": "Rename 'provider' to 'cloud.provider'",
                "field": "provider",
                "target_field": "cloud.provider",
                "ignore_failure": True
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        rename: {
          description: "Rename 'provider' to 'cloud.provider'",
          field: 'provider',
          target_field: 'cloud.provider',
          ignore_failure: true
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      rename: {
        description: "Rename 'provider' to 'cloud.provider'",
        field: "provider",
        target_field: "cloud.provider",
        ignore_failure: true,
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "rename": {
        "description": "Rename 'provider' to 'cloud.provider'",
        "field": "provider",
        "target_field": "cloud.provider",
        "ignore_failure": true
      }
    }
  ]
}

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

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "rename": {
                "description": "Rename 'provider' to 'cloud.provider'",
                "field": "provider",
                "target_field": "cloud.provider",
                "on_failure": [
                    {
                        "set": {
                            "description": "Set 'error.message'",
                            "field": "error.message",
                            "value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
                            "override": False
                        }
                    }
                ]
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        rename: {
          description: "Rename 'provider' to 'cloud.provider'",
          field: 'provider',
          target_field: 'cloud.provider',
          on_failure: [
            {
              set: {
                description: "Set 'error.message'",
                field: 'error.message',
                value: "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
                override: false
              }
            }
          ]
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      rename: {
        description: "Rename 'provider' to 'cloud.provider'",
        field: "provider",
        target_field: "cloud.provider",
        on_failure: [
          {
            set: {
              description: "Set 'error.message'",
              field: "error.message",
              value:
                "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
              override: false,
            },
          },
        ],
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "rename": {
        "description": "Rename 'provider' to 'cloud.provider'",
        "field": "provider",
        "target_field": "cloud.provider",
        "on_failure": [
          {
            "set": {
              "description": "Set 'error.message'",
              "field": "error.message",
              "value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
              "override": false
            }
          }
        ]
      }
    }
  ]
}

Вложенные списки обработчиков on_failure позволяют организовать обработку ошибок.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "rename": {
                "description": "Rename 'provider' to 'cloud.provider'",
                "field": "provider",
                "target_field": "cloud.provider",
                "on_failure": [
                    {
                        "set": {
                            "description": "Set 'error.message'",
                            "field": "error.message",
                            "value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
                            "override": False,
                            "on_failure": [
                                {
                                    "set": {
                                        "description": "Set 'error.message.multi'",
                                        "field": "error.message.multi",
                                        "value": "Document encountered multiple ingest errors",
                                        "override": True
                                    }
                                }
                            ]
                        }
                    }
                ]
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        rename: {
          description: "Rename 'provider' to 'cloud.provider'",
          field: 'provider',
          target_field: 'cloud.provider',
          on_failure: [
            {
              set: {
                description: "Set 'error.message'",
                field: 'error.message',
                value: "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
                override: false,
                on_failure: [
                  {
                    set: {
                      description: "Set 'error.message.multi'",
                      field: 'error.message.multi',
                      value: 'Document encountered multiple ingest errors',
                      override: true
                    }
                  }
                ]
              }
            }
          ]
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      rename: {
        description: "Rename 'provider' to 'cloud.provider'",
        field: "provider",
        target_field: "cloud.provider",
        on_failure: [
          {
            set: {
              description: "Set 'error.message'",
              field: "error.message",
              value:
                "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
              override: false,
              on_failure: [
                {
                  set: {
                    description: "Set 'error.message.multi'",
                    field: "error.message.multi",
                    value: "Document encountered multiple ingest errors",
                    override: true,
                  },
                },
              ],
            },
          },
        ],
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "rename": {
        "description": "Rename 'provider' to 'cloud.provider'",
        "field": "provider",
        "target_field": "cloud.provider",
        "on_failure": [
          {
            "set": {
              "description": "Set 'error.message'",
              "field": "error.message",
              "value": "Field 'provider' does not exist. Cannot rename to 'cloud.provider'",
              "override": false,
              "on_failure": [
                {
                  "set": {
                    "description": "Set 'error.message.multi'",
                    "field": "error.message.multi",
                    "value": "Document encountered multiple ingest errors",
                    "override": true
                  }
                }
              ]
            }
          }
        ]
      }
    }
  ]
}

Вы также можете указать параметр on_failure для конвейера. Если обработчик без значения on_failure завершается ошибкой, Elasticsearch использует этот параметр уровня конвейера в качестве резервного варианта. Elasticsearch не будет пытаться выполнить оставшиеся обработчики конвейера.

PUT _ingest/pipeline/my-pipeline
{
  "processors": [ ... ],
  "on_failure": [
    {
      "set": {
        "description": "Index document to 'failed-<index>'",
        "field": "_index",
        "value": "failed-{{{ _index }}}"
      }
    }
  ]
}

Дополнительная информация о сбое конвейера может быть доступна в метаданных документа, в полях on_failure_message, on_failure_processor_type, on_failure_processor_tag и on_failure_pipeline. Эти поля доступны только внутри блока on_failure.

Следующий пример использует поля метаданных для включения информации о сбоях конвейера в документы.

PUT _ingest/pipeline/my-pipeline
{
  "processors": [ ... ],
  "on_failure": [
    {
      "set": {
        "description": "Record error information",
        "field": "error_information",
        "value": "Processor {{ _ingest.on_failure_processor_type }} with tag {{ _ingest.on_failure_processor_tag }} in pipeline {{ _ingest.on_failure_pipeline }} failed with message {{ _ingest.on_failure_message }}"
      }
    }
  ]
}

Условное выполнение обработчика

Каждый обработчик поддерживает необязательное условие if, написанное как скрипт Painless. Если оно указано, обработчик выполняется только тогда, когда условие if равно true.

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

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "drop": {
                "description": "Drop documents with 'network.name' of 'Guest'",
                "if": "ctx?.network?.name == 'Guest'"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        drop: {
          description: "Drop documents with 'network.name' of 'Guest'",
          if: "ctx?.network?.name == 'Guest'"
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      drop: {
        description: "Drop documents with 'network.name' of 'Guest'",
        if: "ctx?.network?.name == 'Guest'",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "drop": {
        "description": "Drop documents with 'network.name' of 'Guest'",
        "if": "ctx?.network?.name == 'Guest'"
      }
    }
  ]
}

Если настройка кластера script.painless.regex.enabled включена, в скриптах условия if можно использовать регулярные выражения. Поддерживаемый синтаксис см. в Painless regular expressions.

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

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "set": {
                "description": "If 'url.scheme' is 'http', set 'url.insecure' to true",
                "if": "ctx.url?.scheme =~ /^http[^s]/",
                "field": "url.insecure",
                "value": True
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'my-pipeline',
  body: {
    processors: [
      {
        set: {
          description: "If 'url.scheme' is 'http', set 'url.insecure' to true",
          if: 'ctx.url?.scheme =~ /^http[^s]/',
          field: 'url.insecure',
          value: true
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      set: {
        description: "If 'url.scheme' is 'http', set 'url.insecure' to true",
        if: "ctx.url?.scheme =~ /^http[^s]/",
        field: "url.insecure",
        value: true,
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "set": {
        "description": "If 'url.scheme' is 'http', set 'url.insecure' to true",
        "if": "ctx.url?.scheme =~ /^http[^s]/",
        "field": "url.insecure",
        "value": true
      }
    }
  ]
}

Условия if должны быть указаны как допустимый JSON на одной строке. Однако, вы можете использовать синтаксис тройных кавычек консоли Kibana для записи и отладки более крупных скриптов.

Если возможно, избегайте использования сложных или ресурсоёмких скриптов условий if. Сложные скрипты условий могут замедлить индексацию.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "drop": {
                "description": "Drop documents that don't contain 'prod' tag",
                "if": "\n            Collection tags = ctx.tags;\n            if(tags != null){\n              for (String tag : tags) {\n                if (tag.toLowerCase().contains('prod')) {\n                  return false;\n                }\n              }\n            }\n            return true;\n        "
            }
        }
    ],
)
print(resp)

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      drop: {
        description: "Drop documents that don't contain 'prod' tag",
        if: "\n            Collection tags = ctx.tags;\n            if(tags != null){\n              for (String tag : tags) {\n                if (tag.toLowerCase().contains('prod')) {\n                  return false;\n                }\n              }\n            }\n            return true;\n        ",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "drop": {
        "description": "Drop documents that don't contain 'prod' tag",
        "if": """
            Collection tags = ctx.tags;
            if(tags != null){
              for (String tag : tags) {
                if (tag.toLowerCase().contains('prod')) {
                  return false;
                }
              }
            }
            return true;
        """
      }
    }
  ]
}

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

resp = client.put_script(
    id="my-prod-tag-script",
    script={
        "lang": "painless",
        "source": "\n      Collection tags = ctx.tags;\n      if(tags != null){\n        for (String tag : tags) {\n          if (tag.toLowerCase().contains('prod')) {\n            return false;\n          }\n        }\n      }\n      return true;\n    "
    },
)
print(resp)

resp1 = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "drop": {
                "description": "Drop documents that don't contain 'prod' tag",
                "if": {
                    "id": "my-prod-tag-script"
                }
            }
        }
    ],
)
print(resp1)

const response = await client.putScript({
  id: "my-prod-tag-script",
  script: {
    lang: "painless",
    source:
      "\n      Collection tags = ctx.tags;\n      if(tags != null){\n        for (String tag : tags) {\n          if (tag.toLowerCase().contains('prod')) {\n            return false;\n          }\n        }\n      }\n      return true;\n    ",
  },
});
console.log(response);

const response1 = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      drop: {
        description: "Drop documents that don't contain 'prod' tag",
        if: {
          id: "my-prod-tag-script",
        },
      },
    },
  ],
});
console.log(response1);

PUT _scripts/my-prod-tag-script
{
  "script": {
    "lang": "painless",
    "source": """
      Collection tags = ctx.tags;
      if(tags != null){
        for (String tag : tags) {
          if (tag.toLowerCase().contains('prod')) {
            return false;
          }
        }
      }
      return true;
    """
  }
}

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "drop": {
        "description": "Drop documents that don't contain 'prod' tag",
        "if": { "id": "my-prod-tag-script" }
      }
    }
  ]
}

Входящие документы часто содержат поля-объекты. Если скрипт обработчика пытается получить доступ к полю, родительский объект которого отсутствует, Elasticsearch возвращает ошибку NullPointerException. Чтобы избежать этих исключений, используйте безопасные для null операторы, такие как ?., и пишите свои скрипты, чтобы они были безопасными для null.

Например, ctx.network?.name.equalsIgnoreCase('Guest') не является безопасным для null. ctx.network?.name может вернуть null. Перепишите скрипт как 'Guest'.equalsIgnoreCase(ctx.network?.name), который безопасен для null, потому что Guest всегда не null.

Если вы не можете переписать скрипт, чтобы он был безопасен для null, добавьте явную проверку на null.

resp = client.ingest.put_pipeline(
    id="my-pipeline",
    processors=[
        {
            "drop": {
                "description": "Drop documents that contain 'network.name' of 'Guest'",
                "if": "ctx.network?.name != null && ctx.network.name.contains('Guest')"
            }
        }
    ],
)
print(resp)

const response = await client.ingest.putPipeline({
  id: "my-pipeline",
  processors: [
    {
      drop: {
        description: "Drop documents that contain 'network.name' of 'Guest'",
        if: "ctx.network?.name != null && ctx.network.name.contains('Guest')",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/my-pipeline
{
  "processors": [
    {
      "drop": {
        "description": "Drop documents that contain 'network.name' of 'Guest'",
        "if": "ctx.network?.name != null && ctx.network.name.contains('Guest')"
      }
    }
  ]
}

Условное применение конвейеров

Совместите условие if с обработчиком pipeline, чтобы применять другие конвейеры к документам на основе ваших критериев. Вы можете использовать этот конвейер в качестве конвейера по умолчанию в шаблоне индекса, используемом для настройки нескольких потоков данных или индексов.

resp = client.ingest.put_pipeline(
    id="one-pipeline-to-rule-them-all",
    processors=[
        {
            "pipeline": {
                "description": "If 'service.name' is 'apache_httpd', use 'httpd_pipeline'",
                "if": "ctx.service?.name == 'apache_httpd'",
                "name": "httpd_pipeline"
            }
        },
        {
            "pipeline": {
                "description": "If 'service.name' is 'syslog', use 'syslog_pipeline'",
                "if": "ctx.service?.name == 'syslog'",
                "name": "syslog_pipeline"
            }
        },
        {
            "fail": {
                "description": "If 'service.name' is not 'apache_httpd' or 'syslog', return a failure message",
                "if": "ctx.service?.name != 'apache_httpd' && ctx.service?.name != 'syslog'",
                "message": "This pipeline requires service.name to be either `syslog` or `apache_httpd`"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'one-pipeline-to-rule-them-all',
  body: {
    processors: [
      {
        pipeline: {
          description: "If 'service.name' is 'apache_httpd', use 'httpd_pipeline'",
          if: "ctx.service?.name == 'apache_httpd'",
          name: 'httpd_pipeline'
        }
      },
      {
        pipeline: {
          description: "If 'service.name' is 'syslog', use 'syslog_pipeline'",
          if: "ctx.service?.name == 'syslog'",
          name: 'syslog_pipeline'
        }
      },
      {
        fail: {
          description: "If 'service.name' is not 'apache_httpd' or 'syslog', return a failure message",
          if: "ctx.service?.name != 'apache_httpd' && ctx.service?.name != 'syslog'",
          message: 'This pipeline requires service.name to be either `syslog` or `apache_httpd`'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "one-pipeline-to-rule-them-all",
  processors: [
    {
      pipeline: {
        description:
          "If 'service.name' is 'apache_httpd', use 'httpd_pipeline'",
        if: "ctx.service?.name == 'apache_httpd'",
        name: "httpd_pipeline",
      },
    },
    {
      pipeline: {
        description: "If 'service.name' is 'syslog', use 'syslog_pipeline'",
        if: "ctx.service?.name == 'syslog'",
        name: "syslog_pipeline",
      },
    },
    {
      fail: {
        description:
          "If 'service.name' is not 'apache_httpd' or 'syslog', return a failure message",
        if: "ctx.service?.name != 'apache_httpd' && ctx.service?.name != 'syslog'",
        message:
          "This pipeline requires service.name to be either `syslog` or `apache_httpd`",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/one-pipeline-to-rule-them-all
{
  "processors": [
    {
      "pipeline": {
        "description": "If 'service.name' is 'apache_httpd', use 'httpd_pipeline'",
        "if": "ctx.service?.name == 'apache_httpd'",
        "name": "httpd_pipeline"
      }
    },
    {
      "pipeline": {
        "description": "If 'service.name' is 'syslog', use 'syslog_pipeline'",
        "if": "ctx.service?.name == 'syslog'",
        "name": "syslog_pipeline"
      }
    },
    {
      "fail": {
        "description": "If 'service.name' is not 'apache_httpd' or 'syslog', return a failure message",
        "if": "ctx.service?.name != 'apache_httpd' && ctx.service?.name != 'syslog'",
        "message": "This pipeline requires service.name to be either `syslog` or `apache_httpd`"
      }
    }
  ]
}

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

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

resp = client.nodes.stats(
    metric="ingest",
    filter_path="nodes.*.ingest",
)
print(resp)

response = client.nodes.stats(
  metric: 'ingest',
  filter_path: 'nodes.*.ingest'
)
puts response

const response = await client.nodes.stats({
  metric: "ingest",
  filter_path: "nodes.*.ingest",
});
console.log(response);

GET _nodes/stats/ingest?filter_path=nodes.*.ingest

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