Elasticsearch

The Elasticsearch origin can read data in the following modes:

Batch mode

In batch mode, the origin reads all data returned from the Elasticsearch query, and then the pipeline stops. By default, the origin reads in batch mode.

In batch mode, the origin does not maintain the last-saved offset. When the pipeline stops or when Elasticsearch clears the scroll, the origin does not note where it stops reading. When the origin continues reading after a pipeline restart or after Elasticsearch resets the scroll, the origin processes all requested documents again. This can cause the origin to process duplicate data.

Elasticsearch can clear the scroll for any of the following reasons:

• The scroll timeout value is exceeded.
• Due to a network issue, the pipeline loses the connection to Elasticsearch for more than 30 seconds.
• The pipeline stops when the Delete Scroll on Pipeline Stop property is enabled.

To ensure that the origin does not process duplicate data, configure the origin to run in incremental mode so that the origin does save the offset.

Incremental mode

In incremental mode, the origin performs an initial read of all the data returned from the Elasticsearch query, then the origin periodically runs the query again to check for new data. In incremental mode, the pipeline runs continuously so it can repeat the query at regular intervals.

To use incremental mode, you define the amount of time that the origin waits between queries, and you configure the offset field and initial offset value.

The origin uses an offset field and initial offset value to determine where to start reading data. By default, the offset field is defined as a field named timestamp. You can define any Elasticsearch field as the offset field, though you'll most likely want to use a date field. The initial offset value is a value within the offset field where you want the Elasticsearch origin to start reading.

Include both the offset field and the offset value in the Elasticsearch query.

After the initial read, the origin saves the last read offset value. When the origin runs the next incremental query, the origin uses the last read offset value to continue processing from where it stopped.

In incremental mode, the origin maintains the last-saved offset. When the pipeline stops or when Elasticsearch clears the scroll, the origin notes where it stops reading. When the origin continues reading after a pipeline restart or after Elasticsearch resets the scroll, the origin continues processing from where it stopped. When restarting the pipeline, you can reset the origin to process all requested documents.

Define the query that the origin uses to return data from Elasticsearch. You can define any valid Elasticsearch query.

{
  "query": {
    "match_all": {}
  }
}

You can optionally specify an Elasticsearch index or mapping type to define the scope of the query in either batch or incremental mode.

For example, let's say that you specify the customer index, do not specify a mapping type, configure the origin to use batch mode, and use the default query. The query properties are configured like so:

When you run the pipeline, the query returns all documents within the customer index.

An Elasticsearch query can retrieve large numbers of documents from a single search request. This allows the Elasticsearch origin to run a single query, and then read multiple batches of data from the scroll until no results are left. An Elasticsearch scroll functions like a cursor in a traditional database.

To run an Elasticsearch query using the scroll API, you must set a scroll timeout. The scroll timeout tells Elasticsearch how long it should keep the search context alive. The scroll timeout must be long enough for a single batch to be fully read. When the origin reads another batch from the results, the scroll timeout is reset.

Enter the scroll timeout using Elasticsearch duration time units. For example, enter 1m to set a one minute timeout, or enter 1h to set a one hour timeout.

Optionally, you can configure the origin to delete the scroll when the pipeline stops. When configured, Elasticsearch clears the scroll as soon as the pipeline stops, rather than waiting for the scroll timeout value to be exceeded. Keeping a scroll window open consumes Elasticsearch resources. To free up those resources, configure the origin to clear the scroll when it is no longer being used.

If Elasticsearch clears the scroll because the scroll timeout value is exceeded or because the origin is configured to delete the scroll when the pipeline stops, the origin must run the query again. In batch mode, the origin reads all data returned from the query which can result in duplicate data being processed. In incremental mode, the origin begins reading from the last-saved offset.

The Elasticsearch origin performs parallel processing and enables the creation of a multithreaded pipeline.

The Elasticsearch origin uses multiple concurrent threads based on the Number of Slices property. Each thread connects to the origin system, creates a batch of data, and passes the batch to an available pipeline runner.

A pipeline runner is a sourceless pipeline instance - an instance of the pipeline that includes all of the processors, executors, and destinations in the pipeline and handles all pipeline processing after the origin. Each pipeline runner processes one batch at a time, just like a pipeline that runs on a single thread. When the flow of data slows, the pipeline runners wait idly until they are needed, generating an empty batch at regular intervals. You can configure the Runner Idle Time pipeline property to specify the interval or to opt out of empty batch generation.

Multithreaded pipelines preserve the order of records within each batch, just like a single-threaded pipeline. But since batches are processed by different pipeline runners, the order that batches are written to destinations is not ensured.

For example, the origin runs an Elasticsearch query that returns a large number of documents. To decrease processing time, you want to split the scroll into multiple slices and process each slice in parallel. You set the Number of Slices property to 5. When you start the pipeline, the origin splits the scroll into five slices, then the origin creates five threads, and Data Collector creates a matching number of pipeline runners. Upon receiving data, the origin passes a batch to each of the pipeline runners for processing.

At any given moment, the five pipeline runners can each process a batch, so this multithreaded pipeline processes up to five batches at a time. When incoming data slows, the pipeline runners sit idle, available for use as soon as the data flow increases.

For more information about multithreaded pipelines, see Multithreaded Pipeline Overview.