[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16669040#comment-16669040
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r229396964
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.record.path.validation.RecordPathValidator;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "put",
"index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
--- End diff --
Sorry, early morning and coffee didn't kick in.
Yeah, those values should be removed and they are now. I moved to just
purge nulls and empty strings because they're semantically meaningless AFAICT
in Lucene and ES since they are truthy-false whether they don't exist, are null
or empty. So all gone now if you specify the record path.
> Create new put processors that use the ElasticSearch client service
> ---
>
> Key: NIFI-5248
> URL: https://issues.apache.org/jira/browse/NIFI-5248
> Project: Apache NiFi
> Issue Type: New Feature
>Reporter: Mike Thomsen
>Assignee: Mike Thomsen
>
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16669041#comment-16669041 ] ASF GitHub Bot commented on NIFI-5248: -- Github user MikeThomsen commented on the issue: https://github.com/apache/nifi/pull/2861 @mattyb149 review changes are in. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16668564#comment-16668564
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r229264742
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.record.path.validation.RecordPathValidator;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "put",
"index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
--- End diff --
The purpose of this capability is to be able to specify bulk operations. In
ElasticSearch, you can specify the whole range of CRUD operations in a single
bulk operation. That's what this is going for. So you can say "add this, this
and this, delete this, upsert that and delete this thing too" in one push.
It has the potential to be quite useful to teams doing a lot of work with
ElasticSearch because they will not have to segment changes into different
flows, but can create a single flow that is able to take actions to be
performed and bulk deliver them to ES through one pipeline.
> Create new put processors that use the ElasticSearch client service
> ---
>
>
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16650384#comment-16650384
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r225216435
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
---
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.JsonPath;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import
org.apache.nifi.processors.elasticsearch.put.FlowFileJsonDescription;
+import org.apache.nifi.processors.elasticsearch.put.JsonProcessingError;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "put",
"index"})
+@CapabilityDescription("This processor puts user-supplied JSON into
ElasticSearch. It does not require a schema.")
+public class PutElasticsearchJson extends AbstractProcessor implements
ElasticSearchRestProcessor {
+
+static final PropertyDescriptor ID_ATTRIBUTE = new
PropertyDescriptor.Builder()
+.name("put-es-json-id-attribute")
+.displayName("ID Attribute")
+.description("The attribute to use for setting the document ID in
ElasticSearch. If the payload is an array, " +
+"and this option is used for getting the ID, an exception
will be raised.")
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+.addValidator(Validator.VALID)
+.build();
+static final PropertyDescriptor ID_JSON_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-json-id-json-path")
+.displayName("ID JSONPath")
+.description("If set, the document ID will be pulled from each
json block using this JSONPath operation.")
--- End diff --
Nitpick, "json" should be "JSON" everywhere in the doc for consistency
> Create new put processors that use the ElasticSearch client service
> ---
>
> Key: NIFI-5248
> URL:
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16650383#comment-16650383
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r225208197
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.record.path.validation.RecordPathValidator;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "put",
"index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
--- End diff --
If this is specified, does the field remain in the document? What if the
user didn't want to have that field included in the document, but needed it in
order to specify the operation as a field?
What if the index operation were in an attribute versus a record field?
Like if you partitioned records based on whether they were upsert vs delete
operations (for example). Perhaps consider something like UpdateRecord's
"Replacement Value Strategy" property, which allows you to choose whether the
Operation field would be evaluated as a Record Path or a Literal Value. Both
can support EL, but the latter would allow the setting of a single operation
for all the records in the flow file.
> Create new put processors that use the
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16650381#comment-16650381
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r225215710
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticSearchRestProcessor.java
---
@@ -75,6 +76,19 @@
.required(true)
.build();
+Relationship REL_FAILURE = new Relationship.Builder()
+.name("failure")
+.description("All flowfiles that fail for reasons unrelated to
server availability go to this relationship.")
+.build();
+Relationship REL_RETRY = new Relationship.Builder()
+.name("retry")
+.description("All flowfiles that fail due to server/cluster
availability go to this relationship.")
+.build();
+Relationship REL_SUCCESS = new Relationship.Builder()
+.name("success")
+.description("All flowfiles that succeed in being indexed into
ElasticSearch go here.")
--- End diff --
Maybe "transmitted to ElasticSearch"? Unless "indexed" is a generic enough
term that users would know covers upserts, deletes, and any other supported
operation.
> Create new put processors that use the ElasticSearch client service
> ---
>
> Key: NIFI-5248
> URL: https://issues.apache.org/jira/browse/NIFI-5248
> Project: Apache NiFi
> Issue Type: New Feature
>Reporter: Mike Thomsen
>Assignee: Mike Thomsen
>Priority: Major
>
> Two new processors:
> * PutElasticsearchJson - put raw JSON.
> * PutElasticsearchRecord - put records.
> Both of them should support the general bulk load API and be able to do
> things like insert into multiple indexes from one payload.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16650382#comment-16650382
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r225208562
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.record.path.validation.RecordPathValidator;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "put",
"index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-operation-path")
+.displayName("Operation Record Path")
+.description("A record path expression to retrieve index operation
setting from each record. If left blank, " +
+"all operations will be assumed to be index operations.")
+.addValidator(new RecordPathValidator())
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+.build();
+
+static final PropertyDescriptor ID_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-id-path")
+.displayName("ID Record Path")
+.description("A record path expression to retrieve the ID field
for use with
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16650380#comment-16650380 ] ASF GitHub Bot commented on NIFI-5248: -- Github user mattyb149 commented on a diff in the pull request: https://github.com/apache/nifi/pull/2861#discussion_r225215375 --- Diff: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/pom.xml --- @@ -107,6 +146,38 @@ language governing permissions and limitations under the License. --> + +org.apache.maven.plugins +maven-compiler-plugin --- End diff -- I think we talked about this before but can't find it. Does it need to be here? I thought we can inherit the one from the top-level parent POM? > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16650201#comment-16650201 ] ASF GitHub Bot commented on NIFI-5248: -- Github user MikeThomsen commented on the issue: https://github.com/apache/nifi/pull/2861 @mattyb149 anything? > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16636080#comment-16636080 ] ASF GitHub Bot commented on NIFI-5248: -- Github user MikeThomsen commented on the issue: https://github.com/apache/nifi/pull/2861 @mattyb149 should be good to go now. Found the bug and pushed a fix that worked locally. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16634433#comment-16634433 ] ASF GitHub Bot commented on NIFI-5248: -- Github user MikeThomsen commented on the issue: https://github.com/apache/nifi/pull/2861 One of the lookup integration tests is failing. I need to figure out why. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16634144#comment-16634144 ] ASF GitHub Bot commented on NIFI-5248: -- Github user mattyb149 commented on the issue: https://github.com/apache/nifi/pull/2861 Mind rebasing this one? I think you were waiting on the ES LookupService PR anyway, and now it's in :) > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16598748#comment-16598748 ] ASF GitHub Bot commented on NIFI-5248: -- Github user MikeThomsen commented on the issue: https://github.com/apache/nifi/pull/2861 @mattyb149 Made the requested changes. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16598744#comment-16598744
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r214358747
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.addValidator(Validator.VALID)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-operation-path")
+.displayName("A record path expression to retrieve index operation
setting from each record. If left blank, " +
+"all operations will be assumed to be index operations.")
+.addValidator(Validator.VALID)
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+.build();
+
+static final PropertyDescriptor ID_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-id-path")
+.displayName("ID Record Path")
+.description("A record path expression to retrieve the ID field
for use with ElasticSearch. If left blank " +
+"the ID will
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16597457#comment-16597457
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user MikeThomsen commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r214032109
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.JsonPath;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import
org.apache.nifi.processors.elasticsearch.put.FlowFileJsonDescription;
+import org.apache.nifi.processors.elasticsearch.put.JsonProcessingError;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index"})
--- End diff --
Fixed.
> Create new put processors that use the ElasticSearch client service
> ---
>
> Key: NIFI-5248
> URL: https://issues.apache.org/jira/browse/NIFI-5248
> Project: Apache NiFi
> Issue Type: New Feature
>Reporter: Mike Thomsen
>Assignee: Mike Thomsen
>Priority: Major
>
> Two new processors:
> * PutElasticsearchJson - put raw JSON.
> * PutElasticsearchRecord - put records.
> Both of them should support the general bulk load API and be able to do
> things like insert into multiple indexes from one payload.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16578531#comment-16578531
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r209667102
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.addValidator(Validator.VALID)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-operation-path")
+.displayName("A record path expression to retrieve index operation
setting from each record. If left blank, " +
+"all operations will be assumed to be index operations.")
+.addValidator(Validator.VALID)
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+.build();
+
+static final PropertyDescriptor ID_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-id-path")
+.displayName("ID Record Path")
+.description("A record path expression to retrieve the ID field
for use with ElasticSearch. If left blank " +
+"the ID will be
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16578529#comment-16578529
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r209664984
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.JsonPath;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import
org.apache.nifi.processors.elasticsearch.put.FlowFileJsonDescription;
+import org.apache.nifi.processors.elasticsearch.put.JsonProcessingError;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index"})
+@CapabilityDescription("This processor puts user-supplied JSON into
ElasticSearch. It does not require a schema.")
+public class PutElasticsearchJson extends AbstractProcessor implements
ElasticSearchRestProcessor {
+
+static final PropertyDescriptor ID_ATTRIBUTE = new
PropertyDescriptor.Builder()
+.name("put-es-json-id-attribute")
+.displayName("ID Attribute")
+.description("The attribute to use for setting the document ID in
ElasticSearch. If the payload is an array, " +
+"and this option is used for getting the ID, an exception
will be raised.")
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+.addValidator(Validator.VALID)
+.build();
+static final PropertyDescriptor ID_JSON_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-json-id-json-path")
+.displayName("ID JSONPath")
+.description("If set, the document ID will be pulled from each
json block using this JSONPath operation.")
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+.addValidator(Validator.VALID)
+.build();
+static final PropertyDescriptor INDEX_JSON_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-json-index-json-path")
+.displayName("Index
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16578527#comment-16578527
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r209665193
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index", "record"})
--- End diff --
Same thing applies here for the 5x, 6x tags
> Create new put processors that use the ElasticSearch client service
> ---
>
> Key: NIFI-5248
> URL: https://issues.apache.org/jira/browse/NIFI-5248
> Project: Apache NiFi
> Issue Type: New Feature
>Reporter: Mike Thomsen
>Assignee: Mike Thomsen
>Priority: Major
>
> Two new processors:
> * PutElasticsearchJson - put raw JSON.
> * PutElasticsearchRecord - put records.
> Both of them should support the general bulk load API and be able to do
> things like insert into multiple indexes from one payload.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16578530#comment-16578530
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r209665350
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.addValidator(Validator.VALID)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-operation-path")
+.displayName("A record path expression to retrieve index operation
setting from each record. If left blank, " +
--- End diff --
This should be the description, with "Operation RecordPath" or whatever as
the displayName
> Create new put processors that use the ElasticSearch client service
> ---
>
> Key: NIFI-5248
> URL: https://issues.apache.org/jira/browse/NIFI-5248
> Project: Apache NiFi
> Issue Type: New Feature
>Reporter: Mike Thomsen
>Assignee: Mike Thomsen
>Priority: Major
>
> Two new processors:
> * PutElasticsearchJson - put raw JSON.
> *
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16578532#comment-16578532
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r209666126
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.addValidator(Validator.VALID)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-operation-path")
+.displayName("A record path expression to retrieve index operation
setting from each record. If left blank, " +
+"all operations will be assumed to be index operations.")
+.addValidator(Validator.VALID)
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+.build();
+
+static final PropertyDescriptor ID_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-id-path")
+.displayName("ID Record Path")
+.description("A record path expression to retrieve the ID field
for use with ElasticSearch. If left blank " +
+"the ID will be
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16578533#comment-16578533
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r209665697
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
---
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.RecordPathResult;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index", "record"})
+@CapabilityDescription("A record-aware ElasticSearch put processor that
uses the official Elastic REST client libraries.")
+public class PutElasticsearchRecord extends AbstractProcessor implements
ElasticSearchRestProcessor {
+static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+.name("put-es-record-reader")
+.displayName("Record Reader")
+.description("The record reader to use for reading incoming
records from flowfiles.")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.addValidator(Validator.VALID)
+.build();
+
+static final PropertyDescriptor OPERATION_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-operation-path")
+.displayName("A record path expression to retrieve index operation
setting from each record. If left blank, " +
+"all operations will be assumed to be index operations.")
+.addValidator(Validator.VALID)
+.required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+.build();
+
+static final PropertyDescriptor ID_RECORD_PATH = new
PropertyDescriptor.Builder()
+.name("put-es-record-id-path")
+.displayName("ID Record Path")
+.description("A record path expression to retrieve the ID field
for use with ElasticSearch. If left blank " +
+"the ID will be
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[
https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16578502#comment-16578502
]
ASF GitHub Bot commented on NIFI-5248:
--
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2861#discussion_r209663070
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.elasticsearch;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.JsonPath;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchError;
+import org.apache.nifi.elasticsearch.IndexOperationRequest;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import
org.apache.nifi.processors.elasticsearch.put.FlowFileJsonDescription;
+import org.apache.nifi.processors.elasticsearch.put.JsonProcessingError;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"json", "elasticsearch", "5x", "6x", "put", "index"})
--- End diff --
not sure the 5x and 6x tags should be here, perhaps "elasticsearch5" and
"elasticsearch6" but if it's using REST, it should be valid for just about all
ES version (maybe starting with 5)? I can remove on merge if everything else
looks ok, same goes for the Groovy section of the restapi-processors POM (see
my other comment).
> Create new put processors that use the ElasticSearch client service
> ---
>
> Key: NIFI-5248
> URL: https://issues.apache.org/jira/browse/NIFI-5248
> Project: Apache NiFi
> Issue Type: New Feature
>Reporter: Mike Thomsen
>Assignee: Mike Thomsen
>Priority: Major
>
> Two new processors:
> * PutElasticsearchJson - put raw JSON.
> * PutElasticsearchRecord - put records.
> Both of them should support the general bulk load API and be able to do
> things like insert into multiple indexes from one payload.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16565902#comment-16565902 ] ASF GitHub Bot commented on NIFI-5248: -- Github user mattyb149 commented on a diff in the pull request: https://github.com/apache/nifi/pull/2861#discussion_r207013974 --- Diff: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/pom.xml --- @@ -129,6 +168,69 @@ language governing permissions and limitations under the License. --> + +org.codehaus.mojo --- End diff -- Do we need this section here? I thought we enabled Groovy tests from the top-level POM (although it only happens if there is also at least one Java test). If for some reason it's necessary here, I don't think you need/want to add src/main/groovy as source since the community has decided that non-test code in NiFi must be pure Java. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16536378#comment-16536378 ] ASF GitHub Bot commented on NIFI-5248: -- GitHub user MikeThomsen opened a pull request: https://github.com/apache/nifi/pull/2861 NIFI-5248 Added new Elasticsearch json and record processors. Thank you for submitting a contribution to Apache NiFi. In order to streamline the review of the contribution we ask you to ensure the following steps have been taken: ### For all changes: - [ ] Is there a JIRA ticket associated with this PR? Is it referenced in the commit message? - [ ] Does your PR title start with NIFI- where is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character. - [ ] Has your PR been rebased against the latest commit within the target branch (typically master)? - [ ] Is your initial contribution a single, squashed commit? ### For code changes: - [ ] Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder? - [ ] Have you written or updated unit tests to verify your changes? - [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)? - [ ] If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly? - [ ] If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly? - [ ] If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties? ### For documentation related changes: - [ ] Have you ensured that format looks appropriate for the output in which it is rendered? ### Note: Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible. You can merge this pull request into a Git repository by running: $ git pull https://github.com/MikeThomsen/nifi NIFI-5248 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/nifi/pull/2861.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2861 commit 3e86eb373f8e15edb915facb2d65d06311d05943 Author: Mike Thomsen Date: 2018-07-08T18:51:12Z NIFI-5248 Added new Elasticsearch processors. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16499353#comment-16499353 ] Mike Thomsen commented on NIFI-5248: In my opinion, no they cannot be updated. The problem with ElasticSearch is that it has been very unstable in its Java API compared to Solr between v2, v5 and the roadmap beyond v5. The processor bundles didn't really take much of that into account. The processor bundle that is unversioned in its name is primarily for v2. There's a separate, but incomplete, bundle for v5 that uses the transport API (which is deprecated from the client in v6 onward). The new "restapi" processor pack focuses on client services that use the official–and allegedly stable–new high level REST API. It can be used with V5 as a substitute for the transport API, so that means in the long run we can just deprecate the existing bundles in favor of the new bundle once it's feature-complete. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (NIFI-5248) Create new put processors that use the ElasticSearch client service
[ https://issues.apache.org/jira/browse/NIFI-5248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16499309#comment-16499309 ] Sivaprasanna Sethuraman commented on NIFI-5248: --- [~mike.thomsen] Can the existing processors be rather updated/upgraded instead of building new processors? I haven't gone through the code myself but just asking, if it is possible to enhance the existing processors with the API. So many processors focusing on a single component, in this case ElasticSearch, may confuse end users. > Create new put processors that use the ElasticSearch client service > --- > > Key: NIFI-5248 > URL: https://issues.apache.org/jira/browse/NIFI-5248 > Project: Apache NiFi > Issue Type: New Feature >Reporter: Mike Thomsen >Assignee: Mike Thomsen >Priority: Major > > Two new processors: > * PutElasticsearchJson - put raw JSON. > * PutElasticsearchRecord - put records. > Both of them should support the general bulk load API and be able to do > things like insert into multiple indexes from one payload. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
