You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2017/04/19 20:12:32 UTC
nifi git commit: NIFI-3658: Created processor for converting records
between data formats with like schemas
Repository: nifi
Updated Branches:
refs/heads/master 7e7f9a5de -> b93cf7bbd
NIFI-3658: Created processor for converting records between data formats with like schemas
Signed-off-by: Matt Burgess <ma...@apache.org>
NIFI-3658: Incorporated PR review feedback; added counter; clarified documentation
Signed-off-by: Matt Burgess <ma...@apache.org>
This closes #1668
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/b93cf7bb
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/b93cf7bb
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/b93cf7bb
Branch: refs/heads/master
Commit: b93cf7bbdb3daf400845be6538b4eea66be571cc
Parents: 7e7f9a5
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Apr 19 14:57:03 2017 -0400
Committer: Matt Burgess <ma...@apache.org>
Committed: Wed Apr 19 16:12:14 2017 -0400
----------------------------------------------------------------------
.../nifi/processors/standard/ConvertRecord.java | 161 +++++++++++++++++++
.../org.apache.nifi.processor.Processor | 1 +
.../processors/standard/TestConvertRecord.java | 126 +++++++++++++++
.../standard/util/record/MockRecordWriter.java | 27 +++-
.../apache/nifi/serialization/RecordReader.java | 26 +++
5 files changed, 337 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/b93cf7bb/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertRecord.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertRecord.java
new file mode 100644
index 0000000..9a505a2
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertRecord.java
@@ -0,0 +1,161 @@
+/*
+ * 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.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+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.SupportsBatching;
+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.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.StreamCallback;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.RowRecordReaderFactory;
+import org.apache.nifi.serialization.WriteResult;
+
+@EventDriven
+@SupportsBatching
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@SideEffectFree
+@Tags({"convert", "generic", "schema", "json", "csv", "avro", "log", "logs", "freeform", "text"})
+@WritesAttributes({
+ @WritesAttribute(attribute = "mime.type", description = "Sets the mime.type attribute to the MIME Type specified by the Record Writer"),
+ @WritesAttribute(attribute = "record.count", description = "The number of records in the FlowFile")
+})
+@CapabilityDescription("Converts records from one data format to another using configured Record Reader and Record Write Controller Services. "
+ + "The Reader and Writer must be configured with \"matching\" schemas. By this, we mean the schemas must have the same field names. The types of the fields "
+ + "do not have to be the same if a field value can be coerced from one type to another. For instance, if the input schema has a field named \"balance\" of type double, "
+ + "the output schema can have a field named \"balance\" with a type of string, double, or float. If any field is present in the input that is not present in the output, "
+ + "the field will be left out of the output. If any field is specified in the output schema but is not present in the input data/schema, then the field will not be "
+ + "present in the output or will have a null value, depending on the writer.")
+public class ConvertRecord extends AbstractProcessor {
+
+ 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(RowRecordReaderFactory.class)
+ .required(true)
+ .build();
+ 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();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles that are successfully transformed will be routed to this relationship")
+ .build();
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("If a FlowFile cannot be transformed from the configured input format to the configured output format, "
+ + "the unchanged FlowFile 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);
+ return properties;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final RowRecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RowRecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ final RecordSetWriter writer = writerFactory.createWriter(getLogger());
+
+ final AtomicReference<WriteResult> writeResultRef = new AtomicReference<>();
+
+ final FlowFile original = flowFile;
+ try {
+ flowFile = session.write(flowFile, new StreamCallback() {
+ @Override
+ public void process(final InputStream in, final OutputStream out) throws IOException {
+ try (final RecordReader reader = readerFactory.createRecordReader(original, in, getLogger())) {
+
+ final WriteResult writeResult = writer.write(reader.createRecordSet(), out);
+ writeResultRef.set(writeResult);
+
+ } catch (final MalformedRecordException e) {
+ throw new ProcessException("Could not parse incoming data", e);
+ }
+ }
+ });
+ } catch (final ProcessException e) {
+ getLogger().error("Failed to convert {}", new Object[] {flowFile, e});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ final WriteResult writeResult = writeResultRef.get();
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("record.count", String.valueOf(writeResult.getRecordCount()));
+ attributes.put(CoreAttributes.MIME_TYPE.key(), writer.getMimeType());
+ attributes.putAll(writeResult.getAttributes());
+
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ session.transfer(flowFile, REL_SUCCESS);
+ session.adjustCounter("Records Converted", writeResult.getRecordCount(), false);
+ getLogger().info("Successfully converted {} records for {}", new Object[] {writeResult.getRecordCount(), flowFile});
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/b93cf7bb/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 2f2b0cb..3891ee6 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -18,6 +18,7 @@ org.apache.nifi.processors.standard.CompressContent
org.apache.nifi.processors.standard.ControlRate
org.apache.nifi.processors.standard.ConvertCharacterSet
org.apache.nifi.processors.standard.ConvertJSONToSQL
+org.apache.nifi.processors.standard.ConvertRecord
org.apache.nifi.processors.standard.DebugFlow
org.apache.nifi.processors.standard.DetectDuplicate
org.apache.nifi.processors.standard.DistributeLoad
http://git-wip-us.apache.org/repos/asf/nifi/blob/b93cf7bb/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java
new file mode 100644
index 0000000..0dcaeec
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java
@@ -0,0 +1,126 @@
+/*
+ * 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 static org.junit.Assert.assertTrue;
+
+import org.apache.nifi.processors.standard.util.record.MockRecordParser;
+import org.apache.nifi.processors.standard.util.record.MockRecordWriter;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+
+public class TestConvertRecord {
+
+ @Test
+ public void testSuccessfulConversion() throws InitializationException {
+ final MockRecordParser readerService = new MockRecordParser();
+ final MockRecordWriter writerService = new MockRecordWriter("header", false);
+
+ final TestRunner runner = TestRunners.newTestRunner(ConvertRecord.class);
+ runner.addControllerService("reader", readerService);
+ runner.enableControllerService(readerService);
+ runner.addControllerService("writer", writerService);
+ runner.enableControllerService(writerService);
+
+ runner.setProperty(ConvertRecord.RECORD_READER, "reader");
+ runner.setProperty(ConvertRecord.RECORD_WRITER, "writer");
+
+ readerService.addSchemaField("name", RecordFieldType.STRING);
+ readerService.addSchemaField("age", RecordFieldType.INT);
+
+ readerService.addRecord("John Doe", 48);
+ readerService.addRecord("Jane Doe", 47);
+ readerService.addRecord("Jimmy Doe", 14);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(ConvertRecord.REL_SUCCESS, 1);
+ final MockFlowFile out = runner.getFlowFilesForRelationship(ConvertRecord.REL_SUCCESS).get(0);
+
+ out.assertAttributeEquals("record.count", "3");
+ out.assertAttributeEquals("mime.type", "text/plain");
+ out.assertContentEquals("header\nJohn Doe,48\nJane Doe,47\nJimmy Doe,14\n");
+ }
+
+
+ @Test
+ public void testReadFailure() throws InitializationException {
+ final MockRecordParser readerService = new MockRecordParser(2);
+ final MockRecordWriter writerService = new MockRecordWriter("header", false);
+
+ final TestRunner runner = TestRunners.newTestRunner(ConvertRecord.class);
+ runner.addControllerService("reader", readerService);
+ runner.enableControllerService(readerService);
+ runner.addControllerService("writer", writerService);
+ runner.enableControllerService(writerService);
+
+ runner.setProperty(ConvertRecord.RECORD_READER, "reader");
+ runner.setProperty(ConvertRecord.RECORD_WRITER, "writer");
+
+ readerService.addSchemaField("name", RecordFieldType.STRING);
+ readerService.addSchemaField("age", RecordFieldType.INT);
+
+ readerService.addRecord("John Doe", 48);
+ readerService.addRecord("Jane Doe", 47);
+ readerService.addRecord("Jimmy Doe", 14);
+
+ final MockFlowFile original = runner.enqueue("hello");
+ runner.run();
+
+ // Original FlowFile should be routed to 'failure' relationship without modification
+ runner.assertAllFlowFilesTransferred(ConvertRecord.REL_FAILURE, 1);
+ final MockFlowFile out = runner.getFlowFilesForRelationship(ConvertRecord.REL_FAILURE).get(0);
+ assertTrue(original == out);
+ }
+
+
+ @Test
+ public void testWriteFailure() throws InitializationException {
+ final MockRecordParser readerService = new MockRecordParser();
+ final MockRecordWriter writerService = new MockRecordWriter("header", false, 2);
+
+ final TestRunner runner = TestRunners.newTestRunner(ConvertRecord.class);
+ runner.addControllerService("reader", readerService);
+ runner.enableControllerService(readerService);
+ runner.addControllerService("writer", writerService);
+ runner.enableControllerService(writerService);
+
+ runner.setProperty(ConvertRecord.RECORD_READER, "reader");
+ runner.setProperty(ConvertRecord.RECORD_WRITER, "writer");
+
+ readerService.addSchemaField("name", RecordFieldType.STRING);
+ readerService.addSchemaField("age", RecordFieldType.INT);
+
+ readerService.addRecord("John Doe", 48);
+ readerService.addRecord("Jane Doe", 47);
+ readerService.addRecord("Jimmy Doe", 14);
+
+ final MockFlowFile original = runner.enqueue("hello");
+ runner.run();
+
+ // Original FlowFile should be routed to 'failure' relationship without modification
+ runner.assertAllFlowFilesTransferred(ConvertRecord.REL_FAILURE, 1);
+ final MockFlowFile out = runner.getFlowFilesForRelationship(ConvertRecord.REL_FAILURE).get(0);
+ assertTrue(original == out);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/b93cf7bb/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/record/MockRecordWriter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/record/MockRecordWriter.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/record/MockRecordWriter.java
index 1cf2a28..0a57b29 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/record/MockRecordWriter.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/util/record/MockRecordWriter.java
@@ -31,9 +31,21 @@ import org.apache.nifi.serialization.record.RecordSet;
public class MockRecordWriter extends AbstractControllerService implements RecordSetWriterFactory {
private final String header;
+ private final int failAfterN;
+ private final boolean quoteValues;
public MockRecordWriter(final String header) {
+ this(header, true, -1);
+ }
+
+ public MockRecordWriter(final String header, final boolean quoteValues) {
+ this(header, quoteValues, -1);
+ }
+
+ public MockRecordWriter(final String header, final boolean quoteValues, final int failAfterN) {
this.header = header;
+ this.quoteValues = quoteValues;
+ this.failAfterN = failAfterN;
}
@Override
@@ -48,13 +60,20 @@ public class MockRecordWriter extends AbstractControllerService implements Recor
final int numCols = rs.getSchema().getFieldCount();
Record record = null;
while ((record = rs.next()) != null) {
- recordCount++;
+ if (++recordCount > failAfterN && failAfterN > -1) {
+ throw new IOException("Unit Test intentionally throwing IOException after " + failAfterN + " records were written");
+ }
+
int i = 0;
for (final String fieldName : record.getSchema().getFieldNames()) {
final String val = record.getAsString(fieldName);
- out.write("\"".getBytes());
- out.write(val.getBytes());
- out.write("\"".getBytes());
+ if (quoteValues) {
+ out.write("\"".getBytes());
+ out.write(val.getBytes());
+ out.write("\"".getBytes());
+ } else {
+ out.write(val.getBytes());
+ }
if (i++ < numCols - 1) {
out.write(",".getBytes());
http://git-wip-us.apache.org/repos/asf/nifi/blob/b93cf7bb/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordReader.java
index b728498..add248e 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordReader.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.RecordSet;
/**
* <p>
@@ -51,4 +52,29 @@ public interface RecordReader extends Closeable {
* @throws MalformedRecordException if an unrecoverable failure occurs when trying to parse the underlying data
*/
RecordSchema getSchema() throws MalformedRecordException;
+
+ /**
+ * @return a RecordSet that returns the records in this Record Reader in a streaming fashion
+ */
+ default RecordSet createRecordSet() {
+ return new RecordSet() {
+ @Override
+ public RecordSchema getSchema() throws IOException {
+ try {
+ return RecordReader.this.getSchema();
+ } catch (final MalformedRecordException mre) {
+ throw new IOException(mre);
+ }
+ }
+
+ @Override
+ public Record next() throws IOException {
+ try {
+ return RecordReader.this.nextRecord();
+ } catch (final MalformedRecordException mre) {
+ throw new IOException(mre);
+ }
+ }
+ };
+ }
}