Apache NiFi and Elasticsearch integration

How Apache NiFi indexes and queries Elasticsearch

Operational records often need to reach Elasticsearch in a searchable form, and sometimes the results of a search need to flow back into a NiFi pipeline for enrichment, routing or archiving. In NiFi 2, the supported way to do this is the Elasticsearch processor bundle. PutElasticsearchRecord handles writes: it reads FlowFiles through a record reader and sends their records to an index in bulk. The JSON query and consume processors handle reads, running a query against an index and returning hits as FlowFiles. Both directions use an Elasticsearch connection controller service, so credentials, TLS settings and endpoints are defined once and reused. In a DeployOpen engagement, the NiFi cluster, the connection to the Elasticsearch cluster and the credentials all sit inside the client's own network, cloud accounts or data center, and the flow is designed and tested there.

The integration does not decide what the data should look like in the index. Field mappings, analyzers, index templates, shard counts and the index lifecycle policies that roll indices over and delete old data remain the responsibility of the team that owns the Elasticsearch cluster. NiFi can pass a mapping mismatch or a rejected document back as a failure relationship, but it will not redesign the schema for you. Retention, access to indices and document-level security are also configured in Elasticsearch and in your identity setup, not in the flow. Before building, confirm the exact processor names, their property names and the Elasticsearch versions each NiFi release supports against the component documentation for the NiFi release you plan to run, because these details change between releases and the documentation is the authoritative source for them.

What the Elasticsearch integration needs before the first flow

Most Elasticsearch delivery problems in NiFi trace back to decisions made before the first processor is placed. The items below cover the runtime, the cluster, the identity NiFi uses, the network and trust path, and the sign-off on what gets written where.

NiFi release and Java runtime

Confirm the NiFi version you will run and the Java runtime beneath it before choosing processor versions. NiFi 2 requires Java 21 as the minimum runtime, and the NiFi 1.x line is no longer maintained, so an Elasticsearch flow built there will need rework during a later migration. Record the exact release, the JDK build, and the operating system image each node uses. Ask whether the NiFi build is pinned in your deployment tooling so every node runs the same version.

Elasticsearch cluster and version

Confirm the Elasticsearch version and deployment model against the processor documentation for the NiFi release you chose, because supported versions differ between processor bundles. Identify whether you write to a self-managed cluster, a managed service, or a cluster reached through a proxy. Ask the search team which cluster owns the target indexes, which nodes accept bulk traffic, and whether ingest pipelines or index templates already exist that NiFi output must match.

Service account and index scope

Create a dedicated Elasticsearch user for NiFi with privileges limited to the indexes and operations the flow needs, such as bulk writes to named indexes and reads for query processors. Avoid reusing an administrator account or a shared search user. Confirm whether the flow also needs to create indexes or only write to existing ones, since that choice determines the privileges granted. Store the credential in a NiFi sensitive property or parameter, never in a flow file.

Network paths from every node

In a clustered NiFi, every node that runs an Elasticsearch processor opens its own connections, so the network path must exist from each node, not only from the primary node. Verify routing, firewall rules, DNS resolution and any load balancer or proxy between NiFi and the cluster for each node address. Ask the network team to confirm the ports and whether outbound traffic passes through an egress control that logs or inspects requests.

TLS chain and trust store

Obtain the certificate chain that Elasticsearch presents, including any internal certificate authority, and load it into the trust store that the NiFi SSL context service uses. Test the chain from each node, since a node with an outdated trust store fails while others succeed. Confirm certificate expiry dates and the renewal owner, and whether NiFi must present a client certificate. Keep the trust store file and its password under the same change control as the rest of the NiFi configuration.

Approvals for indexes, mappings and retention

Before the build starts, name who approves index names, field mappings, and retention periods. The data owner decides which fields are indexed and how long documents stay searchable, the search or platform team owns mappings and lifecycle policies, and the security owner signs off on any sensitive fields reaching the cluster. Ask for these approvals in writing, because changing a mapping after documents are indexed usually means reindexing.

Configuring the Elasticsearch connection and first index flow

This sequence takes a NiFi flow from an empty Elasticsearch connection to a verified first index, with each step closing on a check that proves the connection, schema, operation and batching behave as intended before the schedule is enabled.

Confirm Elasticsearch version and target index design
Before touching NiFi, confirm with the Elasticsearch owner which major version the cluster runs and whether the NiFi processor bundle you deploy supports it. Agree the index name pattern, the field mappings for the first document type, and whether the index will be written through an alias. Record these decisions in the flow's change ticket. Done means you have a written version, index name and mapping owner, and a test index can be created on that cluster by an account you control.
Create the Elasticsearch controller service
In NiFi, add the Elasticsearch client controller service at the root or process group level, set the cluster endpoint URLs, and choose the authentication method your cluster requires, such as an API key or username and password held as sensitive values. Use TLS with the trust store your security team provides rather than disabling certificate checks. Enable the service and confirm it reports valid. The check is a clean validation state plus a successful health or info request against a non-production cluster.
Configure the record reader to match the source schema
Add a record reader that matches the incoming format, such as JsonTreeReader for JSON FlowFiles, and attach an explicit schema through Schema Access Strategy instead of relying on inference for production data. Compare the schema fields with the Elasticsearch mapping for the target index so type mismatches, such as strings sent to numeric fields, surface now. The check is a sample FlowFile from the upstream processor that parses with zero record errors in the reader's test or in a dry run through a funnel.
Set index and per-document operation
On PutElasticsearchRecord, set the Index name, the Index Operation (index, update, upsert or delete), and the Id Record Path that points at the document identifier in each record. Use an expression that reads the index from a FlowFile attribute only if the attribute is populated and validated upstream. For update and upsert, confirm which fields the operation writes and that the id field is present on every record. The check is a configuration review in which every record in the sample carries a non-empty identifier and the operation matches the design agreed in step one.
Tune batch size and bulk settings
Set the batch size and the bulk request limits on the processor to match the cluster's indexing capacity, which you should confirm with the Elasticsearch team and not guess. Start small for the first run so a rejected bulk request affects a limited number of records. Decide how failed records are routed, typically to a failure relationship with the error attributes preserved for replay. The check is a documented set of values in the change ticket plus a failure path that is connected to a visible sink, not left unconnected.
Run a single test FlowFile into a test index
Point the processor at a non-production test index, feed one FlowFile with a known record, and run the processor once by starting it manually or with a single trigger. Do not enable the schedule yet. Inspect the processor's success and failure counts and the provenance event for the FlowFile. The check is that the document count in the test index rises by exactly one, and that the mapping created or used for the field types matches the design from step one.
Verify the document and mapping in Elasticsearch
Query the test index with a simple search for the known document identifier and confirm the returned source contains the expected fields and values. Inspect the index mapping to confirm that field types are what you designed, not dynamically inferred defaults that would break later queries. Repeat one update or upsert on the same identifier to confirm the operation behaves as specified. The check is a search hit with the expected content and a mapping review signed off by the index owner before the schedule is enabled.

How record fields map to Elasticsearch documents

This table lists the mapping decisions that shape each Elasticsearch document written by PutElasticsearchRecord, along with the team that should own each one. Read it before the first indexing run, because index names, document IDs and field types are hard to change once data is in the index.

Decision	What it controls	Owner	Confirm before go-live
Source record schema	Which fields the record reader passes on and their data types, such as string, long or timestamp	Integration engineer with the source system owner	The reader schema matches the source contract and rejects unexpected types
Index name via record path	Whether each document goes to a fixed index or one derived from a field, such as a tenant or date value	Search team, with data governance for retention	Index naming convention and the maximum number of indices the cluster will hold
Document ID via record path	The _id used for upserts and overwrites, which decides whether reruns create duplicates	Integration engineer and search team	The ID field is unique, stable and present on every record
Operation type per document	Whether each record is indexed, created, updated, upserted or deleted, through the operation record path	Integration engineer	Delete handling is explicitly agreed, since deletes remove data from search
Date formats	How timestamp strings are parsed into date fields, including time zone	Search team with the source system owner	A single agreed format and time zone for each date field in the index mapping
Keyword versus text fields	Whether values are exact-match filters and aggregations or full-text search	Search team, with analysts who run queries	Each field that is filtered, sorted or aggregated is mapped as keyword or a numeric type
Null and missing values	Whether nulls are stored, omitted or replaced with defaults, and how this affects queries	Search team and data owner	Queries treat missing values as agreed, and null-handling rules are documented per field
Mapping changes over time	How new fields and type changes are introduced without breaking existing indices	Search team, with platform operations	A change process exists that avoids reindexing surprises in production

Bulk indexing or per-document operations: choosing the write pattern

Bulk batching sends many documents to Elasticsearch in one request, which reduces request overhead per document. In PutElasticsearchRecord, the record reader splits an incoming FlowFile into records, and the processor groups them into bulk requests. The main levers are how many records each request carries and how much data NiFi holds in memory before it sends. Larger batches usually raise throughput, but they increase heap use and let one bad document affect more of the batch.

Batch size: Size batches from measured document size and heap headroom on each NiFi node, using production-like test data.
Memory footprint: Each batch stays in NiFi memory until the request completes, so concurrent tasks multiply that cost.
Cluster load: Many nodes sending large bulk requests at once can saturate the Elasticsearch ingest path, so agree a concurrency limit with the search team.
Uniform targets: When every record in a batch shares one index and operation, the request is simpler to tune and monitor.

Check batch settings against a test run with realistic document sizes, and watch NiFi heap and Elasticsearch indexing latency together.

Securing the Elasticsearch connection and the indexed data

The NiFi flow holds credentials that can write to or read from your search cluster, and the records it sends often carry the same sensitivity as the source database. The cards below cover how to scope the Elasticsearch identity, store and rotate its secret, verify TLS, control who edits the connection, and keep sensitive values out of logs and provenance. Each control has to match the access model of the source data.

Scope the Elasticsearch role to indexes

Create a dedicated Elasticsearch user for NiFi with a role that grants only the index patterns and operations the flow needs, such as write and create_index on a named index or alias. Avoid superuser or cluster-wide roles, which let a compromised flow read or delete every index. Separate write roles from query roles when the same cluster serves both PutElasticsearchRecord and a JSON query processor. Confirm the exact privilege names with your Elasticsearch version's security documentation before you write the role.

Store the API credential as a sensitive value

Put the Elasticsearch API key or password into a sensitive property of the controller service, or reference it from an external secret source through a parameter. Sensitive values are encrypted in the flow definition and are not shown in the UI after entry. Keep the secret out of processor property values, flow JSON in Git, and parameter contexts that any editor can export. Plan a rotation step that updates the secret in one place and restarts or re-enables the service.

Verify TLS against a managed trust store

Configure the Elasticsearch controller service with the cluster CA in a trust store that NiFi reads, and keep hostname verification enabled. Disabling verification to make a test connection often survives into production and lets a spoofed endpoint receive credentials and documents. Record who owns the CA bundle, how renewals reach every NiFi node, and what happens when a certificate expires mid-batch. Test each node individually, since a clustered flow can fail on one node only.

Restrict who can change the connection

The controller service determines where data goes and with which credential, so changing it is a higher-risk action than editing a processor. Use NiFi's policies to limit the write permission on the controller service and the process group to a small change group, and keep operators to view or start-stop rights. Require a flow change review before the connection URL, index target or credential reference changes. Confirm how your NiFi authorizer maps these policies to your identity provider groups.

Keep sensitive fields out of logs and provenance

Provenance events record FlowFile attributes and content references, and bulletins and logs can capture error text from a failed bulk request. Remove or hash sensitive fields before they reach the indexing processor, and avoid placing secrets or personal identifiers in FlowFile attributes that the flow copies into index names or document IDs. Review who can view provenance and content for this process group, and set retention deliberately. Test a failed bulk call and check what appears in the bulletin board and logs.

Match search access to the source model

A document copied into Elasticsearch can be searched by anyone with read access to its index, even if the source database restricted the same rows by role or tenant. Map each source access rule to an index boundary, a document-level security filter, or a field-level restriction in Elasticsearch before the first load. Confirm with the data owner which fields may be indexed at all, and re-check the mapping when the source schema changes. This is a design decision to settle before indexing starts, not a setting to add afterward.

Testing Elasticsearch indexing before cutover

Verify the PutElasticsearchRecord flow with representative data before any production schedule is enabled. Replay a sample set taken from the real source extract, covering the record shapes that matter most: nested objects, optional fields, dates, and the largest documents. Then compare document counts per index against the source query, and sample documents field by field so that types, timestamps and null handling survive the record reader and writer. Mismatches in field types often show up only in Elasticsearch mappings, so check the index mapping after the first run, not just the document count. Agree the test plan and acceptance criteria with the data owners and the search team before the production schedule is switched on, and record the thresholds in writing so that go or no-go is not decided during the cutover window.

Test duplicate delivery and reruns deliberately. Stop and restart the flow mid-batch, re-run the same input, and confirm that the chosen operation (index, create or upsert with a stable _id derived from the source key) leaves the same document count and content. Where a record path supplies the document ID, verify that the expression resolves for every record type in the sample, because a missing ID can turn an update into a new document. Also test a bulk request that partially fails. Confirm that failed records are routed to the failure relationship, that each one carries its provenance and error attributes, and that a retry path does not double-send the records that already succeeded.

Load test batch sizing on a non-production cluster that uses the same Elasticsearch version, shard layout and heap settings as production, because results on a smaller cluster will not carry over. Increase the record count per FlowFile and the concurrent tasks in steps, watching NiFi back pressure, Elasticsearch bulk rejections and indexing latency. Pick the batch size that stays clear of rejections with headroom for peak load, then document it. Confirm the supported Elasticsearch version range for your NiFi release in the component documentation before you finalize the test matrix, and ask the search team to confirm the cluster's own support status.

Common Elasticsearch integration failures and how to fix them

This table lists the failures teams most often hit when NiFi writes to or reads from Elasticsearch, with the likely cause, the check to run and the fix for each. Use it as a first-pass triage guide, then confirm version-specific behavior in the component documentation for your NiFi and Elasticsearch releases.

Failure	Likely cause	Check and fix
TLS handshake errors	The SSL context service does not trust the Elasticsearch CA, or the HTTP endpoint is configured with the wrong scheme	Confirm the URL scheme, load the CA chain into the SSL context service, and test the connection from the NiFi node that runs the processor
Authentication or role rejections	The API key or user lacks index privileges for the target index or the bulk write action	Check the response code and error text in the failed FlowFile, then grant the minimum write or read privileges on the specific index pattern
Mapping conflicts that reject documents	A field arrives with a type that differs from the index mapping, such as a string in a numeric field	Inspect the rejected document in the failure relationship, align the record schema with the mapping, or create the index with an explicit mapping before loading
Bulk partial failures	Some documents in a batch are rejected while others succeed, so one request returns mixed results	Route per-item failures separately rather than treating the whole batch as sent, and reprocess only the failed records after fixing their cause
Index name errors from record paths	The record path that builds the index name resolves to null, uppercase characters, or characters Elasticsearch does not accept	Add a default or validation step before the processor, lowercase the value, and confirm the resolved name in a test run
Duplicate documents after retries	A retry resends a batch that Elasticsearch already accepted, and no document ID was set	Map a stable business key to the document ID so that re-sent records overwrite rather than duplicate, and confirm the operation type in the component documentation
Query processors returning truncated results	The query size is below the expected hit count, or the consume processor stops at a page or scroll limit	Compare the returned hit count with a count query, adjust the size and pagination settings, and confirm the behavior for your Elasticsearch version in the component documentation

Questions teams ask before an Elasticsearch integration

How long does an Apache NiFi and Elasticsearch integration take?

Duration depends mainly on how many indices and document types the flow must handle, how varied the source records are, and how many per-document operations (index, update, upsert or delete) the design requires. A single index fed from one clean JSON source is a small build. Multiple indices with different mappings, retry rules and dead-letter handling take longer to design, test and hand over. No general duration applies, so ask for a timeline that is tied to a written scope with named sources, targets and acceptance tests.

What drives the effort of an Elasticsearch integration?

Four inputs shape most of the effort:

The number of source formats and how much normalization each one needs before indexing.
The mapping and index design agreed with the search owners, including field types and how dates and nested objects are stored.
Error handling for rejected documents, partial bulk failures and version conflicts.
Authentication, TLS and the cluster topology, which affect connection setup and testing.

Each of these adds design and verification work, so a written scope should list them explicitly.

Can an internal team build an Elasticsearch integration in NiFi on its own?

Yes, if the team already runs NiFi and has someone who understands record readers, record writers, controller services and Elasticsearch mappings. The usual gaps are bulk batching tuned to the cluster, safe handling of failed documents, and testing against the exact Elasticsearch version in use. Teams often bring in a partner for design review, a fixed-scope build or a handover with runbooks, then keep operations in-house. Decide which of these steps your team can own before you start.

What does the search and data owner need to provide?

The integration needs decisions and access from both sides. Search owners should confirm the target indices, index templates, mappings, retention rules, and which operations are allowed (for example, whether delete is permitted). Data owners should confirm source schemas, field meanings, update frequency, and which records are authoritative. Both should name an approver for changes, and someone should own the test data used for acceptance. Without these inputs, the build stops at the design stage.

How does cost split between the one-time build and ongoing operation?

The one-time cost covers requirements, flow design, controller service setup, testing, documentation and handover. Recurring cost covers running NiFi and the Elasticsearch cluster, monitoring, version upgrades, schema changes, and support time when a source or mapping changes. Your own infrastructure and licensing costs are separate from any services fee. Ask a quote to show each item as one-time or recurring, and state who pays for each component.

Which Elasticsearch versions are supported by the NiFi processors?

Compatibility depends on the NiFi release and the Elasticsearch client it bundles, so the answer must come from the NiFi component documentation for your exact NiFi 2 version. Confirm the supported Elasticsearch major and minor versions for that release, and test against your target cluster before go-live. Also check whether your cluster runs in a managed service with API differences. Do not assume support for a version because a neighboring one works.

What should a written quote for this integration include?

Ask for a quote that states the scope in writing: the sources and target indices, the record formats, the bulk and per-document operation rules, error and dead-letter handling, security settings, and the test plan. It should list deliverables such as flow files, documentation and a handover session, plus assumptions about access and environments. Also ask which items are one-time and which recur, what is excluded, and how changes after the agreed scope are handled. DeployOpen plans most engagements as fixed scope, with the scope and timeline agreed in writing before work starts.

What goes wrong most often with Elasticsearch indexing from NiFi?

The most common problems are oversized or poorly tuned bulk requests that overload the cluster, mapping conflicts when a new field type arrives, and silent loss of documents rejected by Elasticsearch. Other issues include credentials stored in plain flow definitions and retries that duplicate documents. Confirm your error routing, idempotent document IDs and monitoring of rejected counts during the design phase, and test each failure path on a non-production cluster.

How DeployOpen scopes and delivers the Elasticsearch integration

DeployOpen plans, builds, tests and documents Apache NiFi and Elasticsearch integrations inside infrastructure you control. Work starts from a written scope that names the flows, Elasticsearch versions, indices and access model before any build begins.

How we scope

Inventory the source systems, record formats and Elasticsearch clusters that the indexing and query flows must touch.
Confirm which Elasticsearch versions your clusters run and check them against the supported versions in the processor documentation.
Agree in writing which bulk operations, per-document actions and query flows are in the fixed-scope deliverables.

How we build

Design the record readers, writers and schema so PutElasticsearchRecord receives predictable documents with stable identifiers.
Tune bulk batch sizes and backpressure against your cluster capacity, then test failure routing for rejected documents.
Build query and consume flows with JSON query processors, and validate pagination and scheduling on a clustered NiFi.

How we hand over

Deliver the flow definitions, controller service configuration and parameter contexts with a documented deployment procedure.
Provide test evidence for indexing, duplicate handling, error paths and restart behavior against your staging cluster.
Document operational runbooks covering index mappings, credential rotation and the checks to run after an Elasticsearch upgrade.

How we support

Ongoing support, upgrades and continuous testing are available as a retainer after handover.
Retainer work can cover NiFi version upgrades, Elasticsearch version changes and reviews of new processor releases.
Ask for a written quote based on the agreed scope, which sets the deliverables, timelines and support terms.