You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by pvillard31 <gi...@git.apache.org> on 2017/07/25 15:45:03 UTC
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
GitHub user pvillard31 opened a pull request:
https://github.com/apache/nifi/pull/2037
NIFI-4227 - add a ForkRecord processor
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-XXXX where XXXX 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/pvillard31/nifi fork-record
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/2037.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 #2037
----
commit 758b1ff2c32034b23e52eb376db16fd02ba5f527
Author: Pierre Villard <pi...@gmail.com>
Date: 2017-07-25T15:41:49Z
NIFI-4227 - add a ForkRecord processor
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2037#discussion_r137839908
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ForkRecord/additionalDetails.html ---
@@ -0,0 +1,365 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>ForkRecord</title>
+
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <p>
+ ForkRecord allows the user to fork a record into multiple records. To do that, the user must specify
+ a <a href="../../../../../html/record-path-guide.html">RecordPath</a> pointing to a field of type
+ ARRAY containing RECORD elements. The generated flow file will contain the records from the specified
+ array. It is also possible to add in each record all the fields of the parent records from the root
+ level to the record element being forked. However it supposes the fields to add are defined in the
+ schema of the Record Writer controller service.
+ </p>
+
+ <h2>Examples</h2>
+
+ <p>
+ To better understand how this Processor works, we will lay out a few examples. For the sake of these examples, let's assume that our input
+ data is JSON formatted and looks like this:
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 1,
+ "name": "John Doe",
+ "address": "123 My Street",
+ "city": "My City",
+ "state": "MS",
+ "zipCode": "11111",
+ "country": "USA",
+ "accounts": [{
+ "id": 42,
+ "balance": 4750.89
+ }, {
+ "id": 43,
+ "balance": 48212.38
+ }]
+},
+{
+ "id": 2,
+ "name": "Jane Doe",
+ "address": "345 My Street",
+ "city": "Her City",
+ "state": "NY",
+ "zipCode": "22222",
+ "country": "USA",
+ "accounts": [{
+ "id": 45,
+ "balance": 6578.45
+ }, {
+ "id": 46,
+ "balance": 34567.21
+ }]
+}]
+</pre>
+</code>
+
+
+ <h3>Example 1 - Fork without parent fields</h3>
+
+ <p>
+ For this case, we want to create one record per <code>account</code> and we don't care about
+ the other fields. We'll set the Record path property to <code>/accounts</code>. The resulting
+ flow file will contain 4 records and will look like (assuming the Record Writer schema is
+ correctly set):
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 42,
+ "balance": 4750.89
+}, {
+ "id": 43,
+ "balance": 48212.38
+}, {
+ "id": 45,
+ "balance": 6578.45
+}, {
+ "id": 46,
+ "balance": 34567.21
+}]
+</pre>
+</code>
+
+
+ <h3>Example 2 - Fork with parent fields</h3>
+
+ <p>
+ Now, if we set the property "Include parent fields" to true, this will recursively include
--- End diff --
No problem @markap14 ;)
I'll work on it and take care of adding sufficient documentation.
Thanks for your feedback.
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2037#discussion_r133908650
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ForkRecord/additionalDetails.html ---
@@ -0,0 +1,365 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>ForkRecord</title>
+
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <p>
+ ForkRecord allows the user to fork a record into multiple records. To do that, the user must specify
+ a <a href="../../../../../html/record-path-guide.html">RecordPath</a> pointing to a field of type
+ ARRAY containing RECORD elements. The generated flow file will contain the records from the specified
+ array. It is also possible to add in each record all the fields of the parent records from the root
+ level to the record element being forked. However it supposes the fields to add are defined in the
+ schema of the Record Writer controller service.
+ </p>
+
+ <h2>Examples</h2>
+
+ <p>
+ To better understand how this Processor works, we will lay out a few examples. For the sake of these examples, let's assume that our input
+ data is JSON formatted and looks like this:
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 1,
+ "name": "John Doe",
+ "address": "123 My Street",
+ "city": "My City",
+ "state": "MS",
+ "zipCode": "11111",
+ "country": "USA",
+ "accounts": [{
+ "id": 42,
+ "balance": 4750.89
+ }, {
+ "id": 43,
+ "balance": 48212.38
+ }]
+},
+{
+ "id": 2,
+ "name": "Jane Doe",
+ "address": "345 My Street",
+ "city": "Her City",
+ "state": "NY",
+ "zipCode": "22222",
+ "country": "USA",
+ "accounts": [{
+ "id": 45,
+ "balance": 6578.45
+ }, {
+ "id": 46,
+ "balance": 34567.21
+ }]
+}]
+</pre>
+</code>
+
+
+ <h3>Example 1 - Fork without parent fields</h3>
+
+ <p>
+ For this case, we want to create one record per <code>account</code> and we don't care about
+ the other fields. We'll set the Record path property to <code>/accounts</code>. The resulting
+ flow file will contain 4 records and will look like (assuming the Record Writer schema is
+ correctly set):
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 42,
+ "balance": 4750.89
+}, {
+ "id": 43,
+ "balance": 48212.38
+}, {
+ "id": 45,
+ "balance": 6578.45
+}, {
+ "id": 46,
+ "balance": 34567.21
+}]
+</pre>
+</code>
+
+
+ <h3>Example 2 - Fork with parent fields</h3>
+
+ <p>
+ Now, if we set the property "Include parent fields" to true, this will recursively include
--- End diff --
In the use case I have that motivated this processor, being able to "extract" each element and "merge" it with the parent fields is really useful. I agree that this could be achieved with the UpdateRecord processor (and the addition of [NIFI-4270](https://issues.apache.org/jira/browse/NIFI-4270)) but it'd require one processor per type of record and can represent a lot of configuration (in my case, I've hundreds of types of records to "extract").
What about an intermediary solution: the include parent fields property does not really make sense since it is up to the writing schema to define what should be in the output. I propose to rename this property in something like "Fork strategy" with two possible values:
- Split: that would create the output you propose. It won't change the structure but will create 1-element array.
- Extract: that would create the current output with the merging of the parent fields (assuming the output schema does contain the parent fields).
What do you think?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2037: NIFI-4227 - add a ForkRecord processor
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on the issue:
https://github.com/apache/nifi/pull/2037
@pvillard31 sorry it's taken me so long to get back to this - it seems to have slipped between the cracks. I did another review and all looks good (except it seems that the description of "Extract Mode" and "Split Mode" were reversed and "Additional Details" confirmed that so i switched it). Did some tests and all works as expected. Definitely a nice addition to the palette! Thanks! +1 merged to master.
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2037#discussion_r133754979
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ForkRecord.java ---
@@ -0,0 +1,293 @@
+/*
+ * 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.standard;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.io.InputStreamCallback;
+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.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.WriteResult;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.type.ArrayDataType;
+
+@SideEffectFree
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"fork", "record", "content", "array", "stream", "event"})
+@CapabilityDescription("This processor allows the user to fork a record into multiple records. The user must specify a RecordPath pointing "
+ + "to a field of type ARRAY containing RECORD elements. The generated flow file will contain the records from the specified array. "
+ + "It is also possible to add in each record all the fields of the parent records from the root level to the record element being "
+ + "forked. However it supposes the fields to add are defined in the schema of the Record Writer controller service. See examples in "
+ + "the additional details documentation of the processor.")
+@WritesAttributes({
+ @WritesAttribute(attribute = "record.count", description = "The merged FlowFile will have a 'record.count' attribute indicating the number of records "
+ + "that were written to the FlowFile."),
+ @WritesAttribute(attribute = "mime.type", description = "The MIME Type indicated by the Record Writer"),
+ @WritesAttribute(attribute = "<Attributes from Record Writer>", description = "Any Attribute that the configured Record Writer returns will be added to the FlowFile.")
+})
+@SeeAlso({QueryRecord.class, SplitRecord.class, PartitionRecord.class, ConvertRecord.class})
+public class ForkRecord extends AbstractProcessor {
+
+ private volatile RecordPathCache recordPathCache = new RecordPathCache(25);
+
+ public static final PropertyDescriptor RECORD_READER = new PropertyDescriptor.Builder()
+ .name("record-reader")
+ .displayName("Record Reader")
+ .description("Specifies the Controller Service to use for reading incoming data")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+ public static final PropertyDescriptor RECORD_WRITER = new PropertyDescriptor.Builder()
+ .name("record-writer")
+ .displayName("Record Writer")
+ .description("Specifies the Controller Service to use for writing out the records")
+ .identifiesControllerService(RecordSetWriterFactory.class)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor RECORD_PATH = new PropertyDescriptor.Builder()
+ .name("record-path")
+ .displayName("Record Path to Array")
+ .description("A RecordPath that points to an array of records that will be forked.")
+ .addValidator(new RecordPathValidator())
+ .expressionLanguageSupported(true)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor INCLUDE_PARENT_FIELDS = new PropertyDescriptor.Builder()
+ .name("include-parent-fields")
+ .displayName("Include Parent Fields")
+ .description("If set to true, all the fields from the root level to the given array will be added as fields of "
+ + "each element of the array to fork.")
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final Relationship REL_FORK = new Relationship.Builder()
+ .name("fork")
+ .description("The FlowFiles containing the forked records will be routed to this relationship")
+ .build();
+ public static final Relationship REL_ORIGINAL = new Relationship.Builder()
+ .name("original")
+ .description("The original FlowFiles will be routed to this relationship")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("In case a FlowFile generates an error during the fork operation, it will be routed to this relationship")
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new ArrayList<>();
+ properties.add(RECORD_READER);
+ properties.add(RECORD_WRITER);
+ properties.add(RECORD_PATH);
+ properties.add(INCLUDE_PARENT_FIELDS);
+ return properties;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_ORIGINAL);
+ relationships.add(REL_FAILURE);
+ relationships.add(REL_FORK);
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String resultPathText = context.getProperty(RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
+ final RecordPath resultRecordPath = recordPathCache.getCompiled(resultPathText);
+
+ final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ final boolean addParentFields = context.getProperty(INCLUDE_PARENT_FIELDS).asBoolean();
+
+ final FlowFile original = flowFile;
+ final FlowFile outFlowFile = session.create(original);
+ final AtomicInteger readCount = new AtomicInteger(0);
+ final AtomicInteger writeCount = new AtomicInteger(0);
+
+ try {
+
+ session.read(flowFile, new InputStreamCallback() {
+ @Override
+ public void process(final InputStream in) throws IOException {
+ try (final RecordReader reader = readerFactory.createRecordReader(original, in, getLogger())) {
+
+ final RecordSchema writeSchema = writerFactory.getSchema(original, reader.getSchema());
+ final OutputStream out = session.write(outFlowFile);
+
+ try (final RecordSetWriter recordSetWriter = writerFactory.createWriter(getLogger(), writeSchema, original, out)) {
+
+ recordSetWriter.beginRecordSet();
+
+ // we read each record of the input flow file
+ Record record;
+ while ((record = reader.nextRecord()) != null) {
+
+ readCount.incrementAndGet();
+
+ // evaluate record path in each record of the flow file
+ Iterator<FieldValue> it = resultRecordPath.evaluate(record).getSelectedFields().iterator();
+
+ while(it.hasNext()) {
+ FieldValue fieldValue = it.next();
+ RecordFieldType fieldType = fieldValue.getField().getDataType().getFieldType();
+
+ // we want to have an array here, nothing else allowed
+ if(fieldType != RecordFieldType.ARRAY) {
+ getLogger().debug("The record path " + resultPathText + " is matching a field "
+ + "of type " + fieldType + " when the type ARRAY is expected.");
+ continue;
+ }
+
+ // we get the type of the elements of the array
+ final ArrayDataType arrayDataType = (ArrayDataType) fieldValue.getField().getDataType();
+ final DataType elementType = arrayDataType.getElementType();
+
+ // we want to have records in the array
+ if(elementType.getFieldType() != RecordFieldType.RECORD) {
+ getLogger().debug("The record path " + resultPathText + " is matching an array field with "
--- End diff --
Same as above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2037#discussion_r133756360
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ForkRecord.java ---
@@ -0,0 +1,293 @@
+/*
+ * 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.standard;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.io.InputStreamCallback;
+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.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.WriteResult;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.type.ArrayDataType;
+
+@SideEffectFree
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"fork", "record", "content", "array", "stream", "event"})
+@CapabilityDescription("This processor allows the user to fork a record into multiple records. The user must specify a RecordPath pointing "
+ + "to a field of type ARRAY containing RECORD elements. The generated flow file will contain the records from the specified array. "
+ + "It is also possible to add in each record all the fields of the parent records from the root level to the record element being "
+ + "forked. However it supposes the fields to add are defined in the schema of the Record Writer controller service. See examples in "
+ + "the additional details documentation of the processor.")
+@WritesAttributes({
+ @WritesAttribute(attribute = "record.count", description = "The merged FlowFile will have a 'record.count' attribute indicating the number of records "
+ + "that were written to the FlowFile."),
+ @WritesAttribute(attribute = "mime.type", description = "The MIME Type indicated by the Record Writer"),
+ @WritesAttribute(attribute = "<Attributes from Record Writer>", description = "Any Attribute that the configured Record Writer returns will be added to the FlowFile.")
+})
+@SeeAlso({QueryRecord.class, SplitRecord.class, PartitionRecord.class, ConvertRecord.class})
+public class ForkRecord extends AbstractProcessor {
+
+ private volatile RecordPathCache recordPathCache = new RecordPathCache(25);
+
+ public static final PropertyDescriptor RECORD_READER = new PropertyDescriptor.Builder()
+ .name("record-reader")
+ .displayName("Record Reader")
+ .description("Specifies the Controller Service to use for reading incoming data")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+ public static final PropertyDescriptor RECORD_WRITER = new PropertyDescriptor.Builder()
+ .name("record-writer")
+ .displayName("Record Writer")
+ .description("Specifies the Controller Service to use for writing out the records")
+ .identifiesControllerService(RecordSetWriterFactory.class)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor RECORD_PATH = new PropertyDescriptor.Builder()
+ .name("record-path")
+ .displayName("Record Path to Array")
+ .description("A RecordPath that points to an array of records that will be forked.")
+ .addValidator(new RecordPathValidator())
+ .expressionLanguageSupported(true)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor INCLUDE_PARENT_FIELDS = new PropertyDescriptor.Builder()
+ .name("include-parent-fields")
+ .displayName("Include Parent Fields")
+ .description("If set to true, all the fields from the root level to the given array will be added as fields of "
+ + "each element of the array to fork.")
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final Relationship REL_FORK = new Relationship.Builder()
+ .name("fork")
+ .description("The FlowFiles containing the forked records will be routed to this relationship")
+ .build();
+ public static final Relationship REL_ORIGINAL = new Relationship.Builder()
+ .name("original")
+ .description("The original FlowFiles will be routed to this relationship")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("In case a FlowFile generates an error during the fork operation, it will be routed to this relationship")
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new ArrayList<>();
+ properties.add(RECORD_READER);
+ properties.add(RECORD_WRITER);
+ properties.add(RECORD_PATH);
+ properties.add(INCLUDE_PARENT_FIELDS);
+ return properties;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_ORIGINAL);
+ relationships.add(REL_FAILURE);
+ relationships.add(REL_FORK);
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String resultPathText = context.getProperty(RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
+ final RecordPath resultRecordPath = recordPathCache.getCompiled(resultPathText);
+
+ final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ final boolean addParentFields = context.getProperty(INCLUDE_PARENT_FIELDS).asBoolean();
+
+ final FlowFile original = flowFile;
+ final FlowFile outFlowFile = session.create(original);
+ final AtomicInteger readCount = new AtomicInteger(0);
+ final AtomicInteger writeCount = new AtomicInteger(0);
+
+ try {
+
+ session.read(flowFile, new InputStreamCallback() {
+ @Override
+ public void process(final InputStream in) throws IOException {
+ try (final RecordReader reader = readerFactory.createRecordReader(original, in, getLogger())) {
+
+ final RecordSchema writeSchema = writerFactory.getSchema(original, reader.getSchema());
+ final OutputStream out = session.write(outFlowFile);
+
+ try (final RecordSetWriter recordSetWriter = writerFactory.createWriter(getLogger(), writeSchema, original, out)) {
+
+ recordSetWriter.beginRecordSet();
+
+ // we read each record of the input flow file
+ Record record;
+ while ((record = reader.nextRecord()) != null) {
+
+ readCount.incrementAndGet();
+
+ // evaluate record path in each record of the flow file
+ Iterator<FieldValue> it = resultRecordPath.evaluate(record).getSelectedFields().iterator();
+
+ while(it.hasNext()) {
+ FieldValue fieldValue = it.next();
+ RecordFieldType fieldType = fieldValue.getField().getDataType().getFieldType();
+
+ // we want to have an array here, nothing else allowed
+ if(fieldType != RecordFieldType.ARRAY) {
+ getLogger().debug("The record path " + resultPathText + " is matching a field "
--- End diff --
In a previous version of this PR, I was throwing an exception and routing to failure. IIRC I changed to this approach in case the pointed array can be empty in a record. In that case, I don't want to fail the whole flow file and go to the next record. I can make this configurable with an additional property though. Does it sound better to you?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/2037
---
[GitHub] nifi issue #2037: NIFI-4227 - add a ForkRecord processor
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on the issue:
https://github.com/apache/nifi/pull/2037
Hey @markap14, I finally found some time to get back on this guy. I added the "split" mode as you suggested with some unit tests and an example in the additional details doc. Let me know if it looks like what you had in mind.
---
[GitHub] nifi issue #2037: NIFI-4227 - add a ForkRecord processor
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on the issue:
https://github.com/apache/nifi/pull/2037
Rebased against master
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2037#discussion_r137839534
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ForkRecord/additionalDetails.html ---
@@ -0,0 +1,365 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>ForkRecord</title>
+
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <p>
+ ForkRecord allows the user to fork a record into multiple records. To do that, the user must specify
+ a <a href="../../../../../html/record-path-guide.html">RecordPath</a> pointing to a field of type
+ ARRAY containing RECORD elements. The generated flow file will contain the records from the specified
+ array. It is also possible to add in each record all the fields of the parent records from the root
+ level to the record element being forked. However it supposes the fields to add are defined in the
+ schema of the Record Writer controller service.
+ </p>
+
+ <h2>Examples</h2>
+
+ <p>
+ To better understand how this Processor works, we will lay out a few examples. For the sake of these examples, let's assume that our input
+ data is JSON formatted and looks like this:
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 1,
+ "name": "John Doe",
+ "address": "123 My Street",
+ "city": "My City",
+ "state": "MS",
+ "zipCode": "11111",
+ "country": "USA",
+ "accounts": [{
+ "id": 42,
+ "balance": 4750.89
+ }, {
+ "id": 43,
+ "balance": 48212.38
+ }]
+},
+{
+ "id": 2,
+ "name": "Jane Doe",
+ "address": "345 My Street",
+ "city": "Her City",
+ "state": "NY",
+ "zipCode": "22222",
+ "country": "USA",
+ "accounts": [{
+ "id": 45,
+ "balance": 6578.45
+ }, {
+ "id": 46,
+ "balance": 34567.21
+ }]
+}]
+</pre>
+</code>
+
+
+ <h3>Example 1 - Fork without parent fields</h3>
+
+ <p>
+ For this case, we want to create one record per <code>account</code> and we don't care about
+ the other fields. We'll set the Record path property to <code>/accounts</code>. The resulting
+ flow file will contain 4 records and will look like (assuming the Record Writer schema is
+ correctly set):
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 42,
+ "balance": 4750.89
+}, {
+ "id": 43,
+ "balance": 48212.38
+}, {
+ "id": 45,
+ "balance": 6578.45
+}, {
+ "id": 46,
+ "balance": 34567.21
+}]
+</pre>
+</code>
+
+
+ <h3>Example 2 - Fork with parent fields</h3>
+
+ <p>
+ Now, if we set the property "Include parent fields" to true, this will recursively include
--- End diff --
@pvillard31 Sorry - I never saw your comment :( I think what you proposed there is good. We would just want to make sure that we have sufficient examples in the 'additionalDetails.html' for the Processor so that users are able to understand how they two options differ.
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2037#discussion_r133754811
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ForkRecord.java ---
@@ -0,0 +1,293 @@
+/*
+ * 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.standard;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.io.InputStreamCallback;
+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.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.WriteResult;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.type.ArrayDataType;
+
+@SideEffectFree
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"fork", "record", "content", "array", "stream", "event"})
+@CapabilityDescription("This processor allows the user to fork a record into multiple records. The user must specify a RecordPath pointing "
+ + "to a field of type ARRAY containing RECORD elements. The generated flow file will contain the records from the specified array. "
+ + "It is also possible to add in each record all the fields of the parent records from the root level to the record element being "
+ + "forked. However it supposes the fields to add are defined in the schema of the Record Writer controller service. See examples in "
+ + "the additional details documentation of the processor.")
+@WritesAttributes({
+ @WritesAttribute(attribute = "record.count", description = "The merged FlowFile will have a 'record.count' attribute indicating the number of records "
+ + "that were written to the FlowFile."),
+ @WritesAttribute(attribute = "mime.type", description = "The MIME Type indicated by the Record Writer"),
+ @WritesAttribute(attribute = "<Attributes from Record Writer>", description = "Any Attribute that the configured Record Writer returns will be added to the FlowFile.")
+})
+@SeeAlso({QueryRecord.class, SplitRecord.class, PartitionRecord.class, ConvertRecord.class})
+public class ForkRecord extends AbstractProcessor {
+
+ private volatile RecordPathCache recordPathCache = new RecordPathCache(25);
+
+ public static final PropertyDescriptor RECORD_READER = new PropertyDescriptor.Builder()
+ .name("record-reader")
+ .displayName("Record Reader")
+ .description("Specifies the Controller Service to use for reading incoming data")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+ public static final PropertyDescriptor RECORD_WRITER = new PropertyDescriptor.Builder()
+ .name("record-writer")
+ .displayName("Record Writer")
+ .description("Specifies the Controller Service to use for writing out the records")
+ .identifiesControllerService(RecordSetWriterFactory.class)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor RECORD_PATH = new PropertyDescriptor.Builder()
+ .name("record-path")
+ .displayName("Record Path to Array")
+ .description("A RecordPath that points to an array of records that will be forked.")
+ .addValidator(new RecordPathValidator())
+ .expressionLanguageSupported(true)
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor INCLUDE_PARENT_FIELDS = new PropertyDescriptor.Builder()
+ .name("include-parent-fields")
+ .displayName("Include Parent Fields")
+ .description("If set to true, all the fields from the root level to the given array will be added as fields of "
+ + "each element of the array to fork.")
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .required(true)
+ .build();
+
+ public static final Relationship REL_FORK = new Relationship.Builder()
+ .name("fork")
+ .description("The FlowFiles containing the forked records will be routed to this relationship")
+ .build();
+ public static final Relationship REL_ORIGINAL = new Relationship.Builder()
+ .name("original")
+ .description("The original FlowFiles will be routed to this relationship")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("In case a FlowFile generates an error during the fork operation, it will be routed to this relationship")
+ .build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new ArrayList<>();
+ properties.add(RECORD_READER);
+ properties.add(RECORD_WRITER);
+ properties.add(RECORD_PATH);
+ properties.add(INCLUDE_PARENT_FIELDS);
+ return properties;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_ORIGINAL);
+ relationships.add(REL_FAILURE);
+ relationships.add(REL_FORK);
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String resultPathText = context.getProperty(RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
+ final RecordPath resultRecordPath = recordPathCache.getCompiled(resultPathText);
+
+ final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ final boolean addParentFields = context.getProperty(INCLUDE_PARENT_FIELDS).asBoolean();
+
+ final FlowFile original = flowFile;
+ final FlowFile outFlowFile = session.create(original);
+ final AtomicInteger readCount = new AtomicInteger(0);
+ final AtomicInteger writeCount = new AtomicInteger(0);
+
+ try {
+
+ session.read(flowFile, new InputStreamCallback() {
+ @Override
+ public void process(final InputStream in) throws IOException {
+ try (final RecordReader reader = readerFactory.createRecordReader(original, in, getLogger())) {
+
+ final RecordSchema writeSchema = writerFactory.getSchema(original, reader.getSchema());
+ final OutputStream out = session.write(outFlowFile);
+
+ try (final RecordSetWriter recordSetWriter = writerFactory.createWriter(getLogger(), writeSchema, original, out)) {
+
+ recordSetWriter.beginRecordSet();
+
+ // we read each record of the input flow file
+ Record record;
+ while ((record = reader.nextRecord()) != null) {
+
+ readCount.incrementAndGet();
+
+ // evaluate record path in each record of the flow file
+ Iterator<FieldValue> it = resultRecordPath.evaluate(record).getSelectedFields().iterator();
+
+ while(it.hasNext()) {
+ FieldValue fieldValue = it.next();
+ RecordFieldType fieldType = fieldValue.getField().getDataType().getFieldType();
+
+ // we want to have an array here, nothing else allowed
+ if(fieldType != RecordFieldType.ARRAY) {
+ getLogger().debug("The record path " + resultPathText + " is matching a field "
--- End diff --
In this case, it seems to me like this indicates an improper configuration of the processor and should probably go to failure, rather than silently skipping the record. At least, I would want to make it optional. Thoughts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2037: NIFI-4227 - add a ForkRecord processor
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2037#discussion_r133768739
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ForkRecord/additionalDetails.html ---
@@ -0,0 +1,365 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>ForkRecord</title>
+
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <p>
+ ForkRecord allows the user to fork a record into multiple records. To do that, the user must specify
+ a <a href="../../../../../html/record-path-guide.html">RecordPath</a> pointing to a field of type
+ ARRAY containing RECORD elements. The generated flow file will contain the records from the specified
+ array. It is also possible to add in each record all the fields of the parent records from the root
+ level to the record element being forked. However it supposes the fields to add are defined in the
+ schema of the Record Writer controller service.
+ </p>
+
+ <h2>Examples</h2>
+
+ <p>
+ To better understand how this Processor works, we will lay out a few examples. For the sake of these examples, let's assume that our input
+ data is JSON formatted and looks like this:
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 1,
+ "name": "John Doe",
+ "address": "123 My Street",
+ "city": "My City",
+ "state": "MS",
+ "zipCode": "11111",
+ "country": "USA",
+ "accounts": [{
+ "id": 42,
+ "balance": 4750.89
+ }, {
+ "id": 43,
+ "balance": 48212.38
+ }]
+},
+{
+ "id": 2,
+ "name": "Jane Doe",
+ "address": "345 My Street",
+ "city": "Her City",
+ "state": "NY",
+ "zipCode": "22222",
+ "country": "USA",
+ "accounts": [{
+ "id": 45,
+ "balance": 6578.45
+ }, {
+ "id": 46,
+ "balance": 34567.21
+ }]
+}]
+</pre>
+</code>
+
+
+ <h3>Example 1 - Fork without parent fields</h3>
+
+ <p>
+ For this case, we want to create one record per <code>account</code> and we don't care about
+ the other fields. We'll set the Record path property to <code>/accounts</code>. The resulting
+ flow file will contain 4 records and will look like (assuming the Record Writer schema is
+ correctly set):
+ </p>
+
+<code>
+<pre>
+[{
+ "id": 42,
+ "balance": 4750.89
+}, {
+ "id": 43,
+ "balance": 48212.38
+}, {
+ "id": 45,
+ "balance": 6578.45
+}, {
+ "id": 46,
+ "balance": 34567.21
+}]
+</pre>
+</code>
+
+
+ <h3>Example 2 - Fork with parent fields</h3>
+
+ <p>
+ Now, if we set the property "Include parent fields" to true, this will recursively include
--- End diff --
In such a case, I would have actually expected the result to have an 'accounts' field that is a 1-element array.
If we wanted to promote that up to the top level, an UpdateRecord processor could be used.
So in general, the way that i would expect the processor to work is to create a copy of the Record that
contains the exact same structure as the original except each element in the array denoted by the
RecordPath would have a single element in the result. For example, if the input looked like:
```
{
"id": 1,
"members": [{
"id": 42,
"name": "John Doe",
"accounts": [
{
"id": 382,
"name": "first account",
"balance": 17.82
}, {
"id": 482,
"name": "other account",
"balance": 182.34
}
]
}, {
"id": 43,
"name": "Jane Doe",
"accounts": [
{
"id": 492,
"name": "yet another account",
"balance": 21.12
}, {
"id": 513,
"name": "final account",
"balance": 142.22
}
]
}]
}
```
Then, if we set the Record Path to `/members`, I would expect output of:
```
[{
"id": 1,
"members": [{
"id": 42,
"name": "John Doe",
"accounts": [
{
"id": 382,
"name": "first account",
"balance": 17.82
}, {
"id": 482,
"name": "other account",
"balance": 182.34
}
]
}]
}, {
"id": 1,
"members": [{
"id": 43,
"name": "Jane Doe",
"accounts": [
{
"id": 492,
"name": "yet another account",
"balance": 21.12
}, {
"id": 513,
"name": "final account",
"balance": 142.22
}
]
}]
}]
```
But if I set the Record Path to `/members[*]/accounts`, then I would expect
4 records to be output, each having a single element in the 'members' array
and each having a single element in the 'accounts' sub-array.
If we took this approach, I think it would make the processor a lot more powerful
because an UpdateRecord processor could be used next if necessary to pair down
the records to the desired fields. This gives us a lot of flexibility with just 1
or 2 processors.
Thoughts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---