Elasticsearch

The Elasticsearch source can read data in the following modes:

Batch mode

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

In batch mode, the source does not maintain the last-saved offset. When the flow stops or when Elasticsearch clears the scroll, the source does not note where it stops reading. When the source continues reading after a flow restart or after Elasticsearch resets the scroll, the source processes all requested documents again. This can cause the source 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 flow loses the connection to Elasticsearch for more than 30 seconds.
• The flow stops when the Delete Scroll on Flow Stop property is enabled.

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

Incremental mode

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

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

The source 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 source to start reading.

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

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

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

Define the query that the source 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 source to use batch mode, and use the default query. The query properties are configured like so:

When you run the flow, 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 source 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 source 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 source to delete the scroll when the flow stops. When configured, Elasticsearch clears the scroll as soon as the flow 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 source 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 source is configured to delete the scroll when the flow stops, the source must run the query again. In batch mode, the source reads all data returned from the query which can result in duplicate data being processed. In incremental mode, the source begins reading from the last-saved offset.

The Elasticsearch source performs parallel processing and enables the creation of a multithreaded flow.

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

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

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

For example, the source 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 flow, the source splits the scroll into five slices, then the source creates five threads, and Data Collector creates a matching number of flow runners. Upon receiving data, the source passes a batch to each of the flow runners for processing.

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

For more information about multithreaded flows, see Multithreaded flow overview.