› › ›

Создание API преобразования

Создаёт преобразование.

Запрос

PUT _transform/<transform_id>

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

Требуются следующие привилегии:

cluster: manage_transform (роль transform_admin предоставляет эту привилегию)
индексы-источники: read, view_index_metadata
целевой индекс: read, create_index, index. Если настроена retention_policy, также требуется привилегия delete.

Описание

Этот API определяет преобразование, которое копирует данные из исходных индексов, преобразует их и сохраняет в целевом индексе, ориентированном на сущности. Если вы выберите метод pivot для своего преобразования, сущности определяются набором полей group_by в объекте pivot. Если вы выберите метод latest, сущности определяются значениями поля unique_key в объекте latest.

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

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

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

Преобразование запоминает роли пользователя, который его создал, и использует те же роли при выполнении. Если эти роли не имеют необходимых привилегий для исходных и целевых индексов, преобразование завершается с ошибкой при попытке выполнить несанкционированные операции. Если вы предоставляете заголовки вторичной авторизации, вместо этого используются данные этих учётных данных.
Для создания преобразования необходимо использовать Kibana или этот API. Не добавляйте преобразование напрямую в индексы .transform-internal* с помощью API индексов Elasticsearch. Если включены функции безопасности Elasticsearch, не предоставляйте пользователям никаких привилегий на индексы .transform-internal*. Если вы использовали преобразования до версии 7.5, также не предоставляйте пользователям никаких привилегий на индексы .data-frame-internal*.

Необходимо выбрать либо метод latest, либо pivot для своего преобразования; оба метода нельзя использовать в одном преобразовании.

Параметры пути

<transform_id>: (Обязательно, строка) Идентификатор преобразования. Этот идентификатор может содержать строчные буквенно-цифровые символы (a-z и 0-9), дефисы и подчёркивания. Он имеет ограничение в 64 символа и должен начинаться и заканчиваться буквенно-цифровыми символами.

Параметры запроса

defer_validation: (Необязательно, логическое значение) Если true, проверки с отложенным выполнением не выполняются. Это поведение может быть желательно, если исходный индекс не существует до создания преобразования.
timeout: (Необязательно, время) Период ожидания ответа. Если ответ не получен до истечения времени ожидания, запрос завершается с ошибкой. По умолчанию 30s.

Тело запроса

description: (Необязательно, строка) Свободное текстовое описание преобразования.

dest

(Обязательно, объект) Назначение для преобразования.

Свойства dest

index: (Обязательно, строка) Индекс назначения для преобразования.

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

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

aliases: (Необязательно, массив объектов) Псевдонимы, которые должен иметь индекс назначения для преобразования. Псевдонимы обрабатываются с использованием сохранённых учетных данных преобразования, что означает использование вторичных учетных данных, предоставленных во время создания (если указаны как первичные, так и вторичные учетные данные).

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

+ Свойства aliases

Детали

alias: (Обязательно, строка) Имя псевдонима.
move_on_creation: (Необязательно, булево) Будет ли индекс назначения единственным индексом в этом псевдониме. Если true, все остальные индексы будут удалены из этого псевдонима перед добавлением индекса назначения в этот псевдоним. По умолчанию false.

pipeline: (Необязательно, строка) Уникальный идентификатор пайплайна обработки.

frequency

(Необязательно, единицы времени) Интервал проверки изменений в исходных индексах при непрерывной работе преобразования. Минимальное значение 1s, максимальное 1h. Значение по умолчанию 1m.

latest

(Обязательно^*, объект) Метод latest преобразует данные, находя последний документ для каждого уникального ключа.

Свойства latest

sort: (Обязательно, строка) Указывает поле даты, используемое для определения последних документов.
unique_key: (Обязательно, массив строк) Указывает массив одного или нескольких полей, используемых для группировки данных.

_meta: (Необязательно, объект) Определяет необязательные метаданные преобразования.

pivot

(Обязательно^*, объект) Метод pivot преобразует данные путём агрегирования и группировки. Эти объекты определяют поля group by и агрегацию для уменьшения данных.

Свойства pivot

aggregations or aggs

(Обязательно, объект) Определяет способ агрегирования сгруппированных данных. В настоящее время поддерживаются следующие агрегации:

group_by

(Обязательно, объект) Определяет способ группировки данных. Можно определить более одной группировки на один свод. В настоящее время поддерживаются следующие типы группировок:

Свойства группировки могут иметь необязательное свойство missing_bucket. Если оно true, документы без значения в соответствующем поле group_by включаются. По умолчанию false.

retention_policy

(Необязательно, объект) Определяет политику хранения для преобразования. Данные, соответствующие определённым критериям, удаляются из индекса назначения.

Свойства retention_policy

time

(Обязательно, объект) Указывает, что преобразование использует поле времени для установки политики хранения. Данные удаляются, если поле времени для политики хранения существует и содержит данные старше max.age.

Свойства time

field: (Обязательно, строка) Поле даты, используемое для вычисления возраста документа. Установите time.field на существующее поле даты.
max_age: (Обязательно, единицы времени) Указывает максимальный возраст документа в индексе назначения. Документы, старше указанного значения, удаляются из индекса назначения.

settings

(Необязательно, объект) Определяет необязательные параметры преобразования.

Свойства settings

align_checkpoints: (Необязательно, булево) Указывает, следует ли оптимизировать интервалы контрольных точек преобразования для повышения производительности. Такая оптимизация может выровнять интервалы контрольных точек с интервалом гистограммы дат, когда гистограмма дат указана в качестве источника группы в конфигурации преобразования. В результате будет выполняться меньше обновлений документов в целевом индексе, что улучшит общую производительность. Значение по умолчанию — true, что означает, что интервалы контрольных точек будут оптимизированы, если это возможно.
dates_as_epoch_millis: (Необязательно, булево) Определяет, должны ли даты в выводе записываться в формате ISO (по умолчанию) или как миллисекунды с начала эпохи. epoch_millis по умолчанию для преобразований, созданных до версии 7.11. Для совместимого вывода установите это значение в true. Значение по умолчанию — false.
deduce_mappings: (Необязательно, булево) Указывает, должен ли трансформер выводить сопоставления целевого индекса из конфигурации трансформера. Значение по умолчанию — true, что означает, что сопоставления целевого индекса будут выведены, если это возможно.
docs_per_second: (Необязательно, число с плавающей точкой) Указывает ограничение на количество входных документов в секунду. Этот параметр ограничивает преобразование, добавляя время ожидания между запросами поиска. Значение по умолчанию — null, что отключает ограничение.
max_page_search_size: (Необязательно, целое число) Определяет начальный размер страницы для составного агрегирования для каждой контрольной точки. Если возникают исключения разрыва цепи, размер страницы динамически корректируется до меньшего значения. Минимальное значение — 10, максимальное — 65,536. Значение по умолчанию — 500.
num_failure_retries: (Необязательно, целое число) Определяет количество попыток восстановления сбоя до того, как задача преобразования будет помечена как failed. Минимальное значение — 0, максимальное — 100. -1 может использоваться для обозначения бесконечности. В этом случае преобразование никогда не прекращает повторные попытки восстановления сбоя. Значение по умолчанию — значение настройки кластера num_transform_failure_retries.
unattended: (Необязательно, булево) Если true, преобразование выполняется в автономном режиме. В автономном режиме преобразование повторяет попытки неограниченно в случае ошибки, что означает, что преобразование никогда не завершается ошибкой. Установка количества повторов, отличного от бесконечного, приводит к ошибке проверки. По умолчанию false.

source

(Обязательно, объект) Источник данных для преобразования.

Свойства source

index: (Обязательно, строка или массив) Индексы-источники для преобразования. Это может быть один индекс, шаблон индекса (например, "my-index-*"), массив индексов (например, ["my-index-000001", "my-index-000002"]) или массив шаблонов индексов (например, ["my-index-*", "my-other-index-*"]). Для удаленных индексов используйте синтаксис "remote_name:index_name".

Если какие-либо индексы находятся в удаленных кластерах, то у узла-мастера и по крайней мере у одного узла преобразования должен быть узел роли remote_cluster_client.
query: (Необязательно, объект) Условие запроса, которое извлекает подмножество данных из исходного индекса. См. Query DSL.
runtime_mappings: (Необязательно, объект) Определения полей времени выполнения поиска, которые могут использоваться преобразованием. Для полей времени выполнения поиска все узлы данных, включая удаленные узлы, должны быть версии 7.12 или более поздней.

sync

(Необязательно, объект) Определяет свойства, необходимые преобразованиям для непрерывной работы.

Свойства sync

time

(Обязательно, объект) Указывает, что преобразование использует поле времени для синхронизации исходных и целевых индексов.

Свойства time

delay: (Необязательно, единицы времени) Задержка времени между текущим временем и последним временем входных данных. Значение по умолчанию — 60s.
field: (Обязательно, строка) Поле даты, используемое для определения новых документов в источнике.

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

Примеры

В следующем преобразовании используется метод pivot:

resp = client.transform.put_transform(
    transform_id="ecommerce_transform1",
    source={
        "index": "kibana_sample_data_ecommerce",
        "query": {
            "term": {
                "geoip.continent_name": {
                    "value": "Asia"
                }
            }
        }
    },
    pivot={
        "group_by": {
            "customer_id": {
                "terms": {
                    "field": "customer_id",
                    "missing_bucket": True
                }
            }
        },
        "aggregations": {
            "max_price": {
                "max": {
                    "field": "taxful_total_price"
                }
            }
        }
    },
    description="Maximum priced ecommerce data by customer_id in Asia",
    dest={
        "index": "kibana_sample_data_ecommerce_transform1",
        "pipeline": "add_timestamp_pipeline"
    },
    frequency="5m",
    sync={
        "time": {
            "field": "order_date",
            "delay": "60s"
        }
    },
    retention_policy={
        "time": {
            "field": "order_date",
            "max_age": "30d"
        }
    },
)
print(resp)

const response = await client.transform.putTransform({
  transform_id: "ecommerce_transform1",
  source: {
    index: "kibana_sample_data_ecommerce",
    query: {
      term: {
        "geoip.continent_name": {
          value: "Asia",
        },
      },
    },
  },
  pivot: {
    group_by: {
      customer_id: {
        terms: {
          field: "customer_id",
          missing_bucket: true,
        },
      },
    },
    aggregations: {
      max_price: {
        max: {
          field: "taxful_total_price",
        },
      },
    },
  },
  description: "Maximum priced ecommerce data by customer_id in Asia",
  dest: {
    index: "kibana_sample_data_ecommerce_transform1",
    pipeline: "add_timestamp_pipeline",
  },
  frequency: "5m",
  sync: {
    time: {
      field: "order_date",
      delay: "60s",
    },
  },
  retention_policy: {
    time: {
      field: "order_date",
      max_age: "30d",
    },
  },
});
console.log(response);

PUT _transform/ecommerce_transform1
{
  "source": {
    "index": "kibana_sample_data_ecommerce",
    "query": {
      "term": {
        "geoip.continent_name": {
          "value": "Asia"
        }
      }
    }
  },
  "pivot": {
    "group_by": {
      "customer_id": {
        "terms": {
          "field": "customer_id",
          "missing_bucket": true
        }
      }
    },
    "aggregations": {
      "max_price": {
        "max": {
          "field": "taxful_total_price"
        }
      }
    }
  },
  "description": "Maximum priced ecommerce data by customer_id in Asia",
  "dest": {
    "index": "kibana_sample_data_ecommerce_transform1",
    "pipeline": "add_timestamp_pipeline"
  },
  "frequency": "5m",
  "sync": {
    "time": {
      "field": "order_date",
      "delay": "60s"
    }
  },
  "retention_policy": {
    "time": {
      "field": "order_date",
      "max_age": "30d"
    }
  }
}

При создании преобразования вы получите следующие результаты:

{
  "acknowledged" : true
}

В следующем преобразовании используется метод latest:

resp = client.transform.put_transform(
    transform_id="ecommerce_transform2",
    source={
        "index": "kibana_sample_data_ecommerce"
    },
    latest={
        "unique_key": [
            "customer_id"
        ],
        "sort": "order_date"
    },
    description="Latest order for each customer",
    dest={
        "index": "kibana_sample_data_ecommerce_transform2"
    },
    frequency="5m",
    sync={
        "time": {
            "field": "order_date",
            "delay": "60s"
        }
    },
)
print(resp)

const response = await client.transform.putTransform({
  transform_id: "ecommerce_transform2",
  source: {
    index: "kibana_sample_data_ecommerce",
  },
  latest: {
    unique_key: ["customer_id"],
    sort: "order_date",
  },
  description: "Latest order for each customer",
  dest: {
    index: "kibana_sample_data_ecommerce_transform2",
  },
  frequency: "5m",
  sync: {
    time: {
      field: "order_date",
      delay: "60s",
    },
  },
});
console.log(response);

PUT _transform/ecommerce_transform2
{
  "source": {
    "index": "kibana_sample_data_ecommerce"
  },
  "latest": {
    "unique_key": ["customer_id"],
    "sort": "order_date"
  },
  "description": "Latest order for each customer",
  "dest": {
    "index": "kibana_sample_data_ecommerce_transform2"
  },
  "frequency": "5m",
  "sync": {
    "time": {
      "field": "order_date",
      "delay": "60s"
    }
  }
}

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