How transform checkpoints work

Each time a transform examines the source indices and creates or updates the destination index, it generates a checkpoint.

If your transform runs only once, there is logically only one checkpoint. If your transform runs continuously, however, it creates checkpoints as it ingests and transforms new source data. The sync property of the transform configures checkpointing by specifying a time field.

To create a checkpoint, the continuous transform:

This checkpoint process involves both search and indexing activity on the cluster. We have attempted to favor control over performance while developing transforms. We decided it was preferable for the transform to take longer to complete, rather than to finish quickly and take precedence in resource consumption. That being said, the cluster still requires enough resources to support both the composite aggregation search and the indexing of its results.

In most cases, it is strongly recommended to use the ingest timestamp of the source indices for syncing the transform. This is the most optimal way for transforms to be able to identify new changes. If your data source follows the ECS standard, you might already have an event.ingested field. In this case, use event.ingested as the sync.time.field property of your transform.

If you don’t have a event.ingested field or it isn’t populated, you can set it by using an ingest pipeline. Create an ingest pipeline either using the ingest pipeline API (like the example below) or via Kibana under Stack Management > Ingest Pipelines. Use a set processor to set the field and associate it with the value of the ingest timestamp.

resp = client.ingest.put_pipeline(
    id="set_ingest_time",
    description="Set ingest timestamp.",
    processors=[
        {
            "set": {
                "field": "event.ingested",
                "value": "{{{_ingest.timestamp}}}"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'set_ingest_time',
  body: {
    description: 'Set ingest timestamp.',
    processors: [
      {
        set: {
          field: 'event.ingested',
          value: '{{{_ingest.timestamp}}}'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "set_ingest_time",
  description: "Set ingest timestamp.",
  processors: [
    {
      set: {
        field: "event.ingested",
        value: "{{{_ingest.timestamp}}}",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/set_ingest_time
{
  "description": "Set ingest timestamp.",
  "processors": [
    {
      "set": {
        "field": "event.ingested",
        "value": "{{{_ingest.timestamp}}}"
      }
    }
  ]
}

After you created the ingest pipeline, apply it to the source indices of your transform. The pipeline adds the field event.ingested to every document with the value of the ingest timestamp. Configure the sync.time.field property of your transform to use the field by using the create transform API for new transforms or the update transform API for existing transforms. The event.ingested field is used for syncing the transform.

Refer to Add a pipeline to an indexing request and Ingest pipelines to learn more about how to use an ingest pipeline.

When the transform runs in continuous mode, it updates the documents in the destination index as new data comes in. The transform uses a set of heuristics called change detection to update the destination index with fewer operations.

In this example, the data is grouped by host names. Change detection detects which host names have changed, for example, host A, C and G and only updates documents with those hosts but does not update documents that store information about host B, D, or any other host that are not changed.

Another heuristic can be applied for time buckets when a date_histogram is used to group by time buckets. Change detection detects which time buckets have changed and only update those.

Failures in transforms tend to be related to searching or indexing. To increase the resiliency of transforms, the cursor positions of the aggregated search and the changed entities search are tracked in memory and persisted periodically.

Checkpoint failures can be categorized as follows:

If the node running the transforms fails, the transform restarts from the most recent persisted cursor position. This recovery process might repeat some of the work the transform had already done, but it ensures data consistency.