Adam Turley created NIFI-16247:
----------------------------------
Summary: PutElasticsearchJson - Support specifying an
Elasticsearch ingest pipeline, including per-document resolution from the
payload
Key: NIFI-16247
URL: https://issues.apache.org/jira/browse/NIFI-16247
Project: Apache NiFi
Issue Type: Improvement
Components: Extensions
Affects Versions: 2.11.0
Environment: containerized rhel8 environment
Reporter: Adam Turley
Assignee: Adam Turley
Right now there's no straightforward way to send documents through an
Elasticsearch ingest pipeline from PutElasticsearchJson. You can set
BULK:pipeline as a dynamic property, but that applies one pipeline to every
document in the FlowFile and there's no way to pick the pipeline based on the
content of each record.
I'd like to add proper pipeline support that works the same way the Index
configuration already does:
* A Pipeline property for a static pipeline name (supports Expression
Language), same idea as the Index property.
* A Pipeline Field property to pull the pipeline from a field in the document,
same as Index Field. It should respect Field Path Mode so the value can come
from a top-level or nested field.
* A Retain Pipeline Field property to control whether that field stays in the
document or gets removed before indexing.
The resolved value gets added to the bulk action header ("pipeline": "...") per
document for index/create/upsert operations. If Pipeline Field is empty or
missing from a document, it falls back to the Pipeline property.
The main thing this gets us is per-document pipeline routing — different
records in the same batch can go through different ingest pipelines based on
their own content, without having to split the flow or run multiple processors.
It also makes it easy to lean on Elasticsearch ingest pipelines (geoip, grok,
enrich, etc.) for index-time processing straight from NiFi.
This shouldn't need any changes to the client service API since the pipeline is
just another bulk action header field.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
