[
https://issues.apache.org/jira/browse/NIFI-4002?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16033704#comment-16033704
]
ASF GitHub Bot commented on NIFI-4002:
--------------------------------------
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1878#discussion_r119729793
--- Diff:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
---
@@ -0,0 +1,559 @@
+/*
+ * 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 okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.nifi.annotation.behavior.EventDriven;
+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.expression.AttributeExpression;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+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.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPath;
+import org.apache.nifi.record.path.util.RecordPathCache;
+import org.apache.nifi.record.path.validation.RecordPathValidator;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.type.ArrayDataType;
+import org.apache.nifi.serialization.record.type.ChoiceDataType;
+import org.apache.nifi.serialization.record.type.MapDataType;
+import org.apache.nifi.serialization.record.type.RecordDataType;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+import org.apache.nifi.util.StringUtils;
+import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.JsonGenerator;
+import org.codehaus.jackson.JsonNode;
+import org.codehaus.jackson.node.ArrayNode;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.math.BigInteger;
+import java.net.MalformedURLException;
+import java.net.URL;
+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.Optional;
+import java.util.Set;
+
+import static org.apache.commons.lang3.StringUtils.trimToEmpty;
+
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@EventDriven
+@Tags({"elasticsearch", "insert", "update", "upsert", "delete", "write",
"put", "http", "record"})
+@CapabilityDescription("Writes the contents of a FlowFile to
Elasticsearch, using the specified parameters such as "
+ + "the index to insert into and the type of the document.")
+public class PutElasticsearchHttpRecord extends
AbstractElasticsearchHttpProcessor {
+
+ public static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+ .description("All FlowFiles that are written to Elasticsearch
are routed to this relationship").build();
+
+ public static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+ .description("All FlowFiles that cannot be written to
Elasticsearch are routed to this relationship").build();
+
+ public static final Relationship REL_RETRY = new
Relationship.Builder().name("retry")
+ .description("A FlowFile is routed to this relationship if the
database cannot be updated but attempting the operation again may succeed")
+ .build();
+
+ static final PropertyDescriptor RECORD_READER = new
PropertyDescriptor.Builder()
+ .name("put-db-record-record-reader")
+ .displayName("Record Reader")
+ .description("Specifies the Controller Service to use for
parsing incoming data and determining the data's schema.")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor ID_RECORD_PATH = new
PropertyDescriptor.Builder()
+ .name("put-es-record-id-path")
+ .displayName("Identifier Record Path")
+ .description("A RecordPath pointing to a field in the
record(s) that contains the identifier for the document. If the Index Operation
is \"index\", "
+ + "this property may be left empty or evaluate to an
empty value, in which case the document's identifier will be "
+ + "auto-generated by Elasticsearch. For all other
Index Operations, the field's value must be non-empty.")
+ .required(false)
+ .addValidator(new RecordPathValidator())
+ .expressionLanguageSupported(true)
+ .build();
+
+ static final PropertyDescriptor INDEX = new
PropertyDescriptor.Builder()
+ .name("put-es-record-index")
+ .displayName("Index")
+ .description("The name of the index to insert into")
+ .required(true)
+ .expressionLanguageSupported(true)
+
.addValidator(StandardValidators.createAttributeExpressionLanguageValidator(
+ AttributeExpression.ResultType.STRING, true))
+ .build();
+
+ static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
+ .name("put-es-record-type")
+ .displayName("Type")
+ .description("The type of this document (used by Elasticsearch
for indexing and searching)")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor INDEX_OP = new
PropertyDescriptor.Builder()
+ .name("put-es-record-index-op")
+ .displayName("Index Operation")
+ .description("The type of the operation used to index (index,
update, upsert, delete)")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+ .defaultValue("index")
+ .build();
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile RecordPathCache recordPathCache;
+
+ private final JsonFactory factory = new JsonFactory();
+
+ static {
+ final Set<Relationship> _rels = new HashSet<>();
+ _rels.add(REL_SUCCESS);
+ _rels.add(REL_FAILURE);
+ _rels.add(REL_RETRY);
+ relationships = Collections.unmodifiableSet(_rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(ES_URL);
+ descriptors.add(PROP_SSL_CONTEXT_SERVICE);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+ descriptors.add(CONNECT_TIMEOUT);
+ descriptors.add(RESPONSE_TIMEOUT);
+ descriptors.add(RECORD_READER);
+ descriptors.add(ID_RECORD_PATH);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(INDEX_OP);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor>
getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @Override
+ protected Collection<ValidationResult>
customValidate(ValidationContext validationContext) {
+ final List<ValidationResult> problems = new
ArrayList<>(super.customValidate(validationContext));
+ // Since Expression Language is allowed for index operation, we
can't guarantee that we can catch
+ // all invalid configurations, but we should catch them as soon as
we can. For example, if the
+ // Identifier Record Path property is empty, the Index Operation
must evaluate to "index".
+ String idPath =
validationContext.getProperty(ID_RECORD_PATH).getValue();
+ String indexOp =
validationContext.getProperty(INDEX_OP).getValue();
+
+ if (StringUtils.isEmpty(idPath)) {
+ switch (indexOp.toLowerCase()) {
+ case "update":
+ case "upsert":
+ case "delete":
+ case "":
+ problems.add(new ValidationResult.Builder()
+ .valid(false)
+ .subject(INDEX_OP.getDisplayName())
+ .explanation("If Identifier Record Path is not
set, Index Operation must evaluate to \"index\"")
+ .build());
+ break;
+ default:
+ break;
+ }
+ }
+ return problems;
+ }
+
+ @OnScheduled
+ public void setup(ProcessContext context) {
+ super.setup(context);
+ recordPathCache = new RecordPathCache(10);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final
ProcessSession session) throws ProcessException {
+
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final RecordReaderFactory readerFactory =
context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+
+ // Authentication
+ final String username =
context.getProperty(USERNAME).evaluateAttributeExpressions(flowFile).getValue();
+ final String password =
context.getProperty(PASSWORD).evaluateAttributeExpressions(flowFile).getValue();
+
+ OkHttpClient okHttpClient = getClient();
+ final ComponentLog logger = getLogger();
+
+ final String baseUrl =
trimToEmpty(context.getProperty(ES_URL).evaluateAttributeExpressions().getValue());
+ final URL url;
+ try {
+ url = new URL((baseUrl.endsWith("/") ? baseUrl : baseUrl +
"/") + "_bulk");
+ } catch (MalformedURLException mue) {
+ // Since we have a URL validator, something has gone very
wrong, throw a ProcessException
+ context.yield();
+ throw new ProcessException(mue);
+ }
+
+ final String index =
context.getProperty(INDEX).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isEmpty(index)) {
+ logger.error("No value for index in for {}, transferring to
failure", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ final String docType =
context.getProperty(TYPE).evaluateAttributeExpressions(flowFile).getValue();
+ String indexOp =
context.getProperty(INDEX_OP).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isEmpty(indexOp)) {
+ logger.error("No Index operation specified for {},
transferring to failure.", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ switch (indexOp.toLowerCase()) {
+ case "index":
+ case "update":
+ case "upsert":
+ case "delete":
+ break;
+ default:
+ logger.error("Index operation {} not supported for {},
transferring to failure.", new Object[]{indexOp, flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ final String id_path =
context.getProperty(ID_RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
+ final RecordPath recordPath = StringUtils.isEmpty(id_path) ? null
: recordPathCache.getCompiled(id_path);
+ final StringBuilder sb = new StringBuilder();
+
+ try (final InputStream in = session.read(flowFile);
+ final RecordReader reader =
readerFactory.createRecordReader(flowFile, in, getLogger())) {
+
+ Record record;
+ while ((record = reader.nextRecord()) != null) {
+
+ final String id;
+ if (recordPath != null) {
+ Optional<FieldValue> idPathValue =
recordPath.evaluate(record).getSelectedFields().findFirst();
+ if (!idPathValue.isPresent() ||
idPathValue.get().getValue() == null) {
+ throw new IdentifierNotFoundException("Identifier
Record Path specified but no value was found, transferring {} to failure.");
+ }
+ id = idPathValue.get().getValue().toString();
+ } else {
+ id = null;
+ }
+
+ // The ID must be valid for all operations except "index".
For that case,
+ // a missing ID indicates one is to be auto-generated by
Elasticsearch
+ if (id == null && !indexOp.equalsIgnoreCase("index")) {
+ throw new IdentifierNotFoundException("Index operation
{} requires a valid identifier value from a flow file attribute, transferring
to failure.");
+ }
+
+ final StringBuilder json = new StringBuilder();
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ JsonGenerator generator = factory.createJsonGenerator(out);
+ writeRecord(record, record.getSchema(), generator);
+ generator.flush();
+ generator.close();
+ json.append(out.toString());
+
+ if (indexOp.equalsIgnoreCase("index")) {
+ sb.append("{\"index\": { \"_index\": \"");
+ sb.append(index);
+ sb.append("\", \"_type\": \"");
+ sb.append(docType);
+ sb.append("\"");
+ if (!StringUtils.isEmpty(id)) {
+ sb.append(", \"_id\": \"");
+ sb.append(id);
+ sb.append("\"");
+ }
+ sb.append("}}\n");
+ sb.append(json);
+ sb.append("\n");
+ } else if (indexOp.equalsIgnoreCase("upsert") ||
indexOp.equalsIgnoreCase("update")) {
+ sb.append("{\"update\": { \"_index\": \"");
+ sb.append(index);
+ sb.append("\", \"_type\": \"");
+ sb.append(docType);
+ sb.append("\", \"_id\": \"");
+ sb.append(id);
+ sb.append("\" }\n");
+ sb.append("{\"doc\": ");
+ sb.append(json);
+ sb.append(", \"doc_as_upsert\": ");
+ sb.append(indexOp.equalsIgnoreCase("upsert"));
+ sb.append(" }\n");
+ } else if (indexOp.equalsIgnoreCase("delete")) {
+ sb.append("{\"delete\": { \"_index\": \"");
+ sb.append(index);
+ sb.append("\", \"_type\": \"");
+ sb.append(docType);
+ sb.append("\", \"_id\": \"");
+ sb.append(id);
+ sb.append("\" }\n");
+ }
+ }
+ } catch (IdentifierNotFoundException infe) {
+ logger.error(infe.getMessage(), new Object[]{flowFile});
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+
+ } catch (final IOException | SchemaNotFoundException |
MalformedRecordException e) {
+ throw new ProcessException("Could not parse incoming data", e);
+ }
+
+ RequestBody requestBody =
RequestBody.create(MediaType.parse("application/json"), sb.toString());
+ final Response getResponse;
+ try {
+ getResponse = sendRequestToElasticsearch(okHttpClient, url,
username, password, "PUT", requestBody);
+ } catch (final Exception e) {
+ logger.error("Routing to {} due to exception: {}", new
Object[]{REL_FAILURE.getName(), e}, e);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ final int statusCode = getResponse.code();
+
+ if (isSuccess(statusCode)) {
+ ResponseBody responseBody = getResponse.body();
+ try {
+ final byte[] bodyBytes = responseBody.bytes();
+
+ JsonNode responseJson = parseJsonResponse(new
ByteArrayInputStream(bodyBytes));
+ boolean errors =
responseJson.get("errors").asBoolean(false);
+ // ES has no rollback, so if errors occur, log them and
route the whole flow file to failure
+ if (errors) {
+ ArrayNode itemNodeArray = (ArrayNode)
responseJson.get("items");
+ if (itemNodeArray.size() > 0) {
+ // All items are returned whether they succeeded
or failed, so iterate through the item array
+ // at the same time as the flow file list, logging
failures accordingly
+ for (int i = itemNodeArray.size() - 1; i >= 0;
i--) {
+ JsonNode itemNode = itemNodeArray.get(i);
+ int status =
itemNode.findPath("status").asInt();
+ if (!isSuccess(status)) {
+ String reason =
itemNode.findPath("//error/reason").asText();
+ logger.error("Failed to insert {} into
Elasticsearch due to {}, transferring to failure",
+ new Object[]{flowFile, reason});
+ }
+ }
+ }
+ session.transfer(flowFile, REL_FAILURE);
+ } else {
+ session.transfer(flowFile, REL_SUCCESS);
--- End diff --
Yep definitely, forgot to add it, will add now
> Add PutElasticsearchHttpRecord processor
> ----------------------------------------
>
> Key: NIFI-4002
> URL: https://issues.apache.org/jira/browse/NIFI-4002
> Project: Apache NiFi
> Issue Type: New Feature
> Components: Extensions
> Reporter: Matt Burgess
> Assignee: Matt Burgess
>
> With the new Record Reader/Writer capabilities, and that the PutElasticsearch
> processors only handle a single document at a time, it would be nice to have
> a "record-aware" PutES processor, where the user could specify a Record
> Reader and for each record in the input, the processor would convert the
> record to JSON and push to an ES cluster. One necessary (but optional)
> property would be a Record Path to a field/value to be used as the document
> identifier (the current PutES processors require an attribute to hold the
> identifier). If the Record Path is not set and the operation is "index", then
> just like the PutES processors, the document identifier will be autogenerated.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
