You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2017/04/24 21:03:49 UTC
[3/7] nifi git commit: NIFI-3682: This closes #1682. Add Schema
Access Strategy and Schema Write Strategy Record Readers and Writers;
bug fixes.
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVReader.java
index 6b06ebf..fb34f8f 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVReader.java
@@ -17,6 +17,7 @@
package org.apache.nifi.csv;
+import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
@@ -26,21 +27,31 @@ import org.apache.commons.csv.CSVFormat;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.schema.access.SchemaAccessStrategy;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.schemaregistry.services.SchemaRegistry;
import org.apache.nifi.serialization.DateTimeUtils;
import org.apache.nifi.serialization.RecordReader;
-import org.apache.nifi.serialization.RowRecordReaderFactory;
-import org.apache.nifi.serialization.SchemaRegistryRecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.SchemaRegistryService;
import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.stream.io.NonCloseableInputStream;
@Tags({"csv", "parse", "record", "row", "reader", "delimited", "comma", "separated", "values"})
@CapabilityDescription("Parses CSV-formatted data, returning each row in the CSV file as a separate record. "
+ "This reader assumes that the first line in the content is the column names and all subsequent lines are "
+ "the values. See Controller Service's Usage for further documentation.")
-public class CSVReader extends SchemaRegistryRecordReader implements RowRecordReaderFactory {
+public class CSVReader extends SchemaRegistryService implements RecordReaderFactory {
+
+ private final AllowableValue headerDerivedAllowableValue = new AllowableValue("csv-header-derived", "Use String Fields From Header",
+ "The first non-comment line of the CSV file is a header line that contains the names of the columns. The schema will be derived by using the "
+ + "column names in the header and assuming that all columns are of type String.");
+ private final SchemaAccessStrategy headerDerivedSchemaStrategy = new CSVHeaderSchemaStrategy();
private volatile CSVFormat csvFormat;
private volatile String dateFormat;
@@ -56,6 +67,7 @@ public class CSVReader extends SchemaRegistryRecordReader implements RowRecordRe
properties.add(DateTimeUtils.TIMESTAMP_FORMAT);
properties.add(CSVUtils.CSV_FORMAT);
properties.add(CSVUtils.VALUE_SEPARATOR);
+ properties.add(CSVUtils.SKIP_HEADER_LINE);
properties.add(CSVUtils.QUOTE_CHAR);
properties.add(CSVUtils.ESCAPE_CHAR);
properties.add(CSVUtils.COMMENT_MARKER);
@@ -73,9 +85,34 @@ public class CSVReader extends SchemaRegistryRecordReader implements RowRecordRe
}
@Override
- public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException {
- final RecordSchema schema = getSchema(flowFile);
- return new CSVRecordReader(in, logger, schema, csvFormat, dateFormat, timeFormat, timestampFormat);
+ public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException, SchemaNotFoundException {
+ // Use Mark/Reset of a BufferedInputStream in case we read from the Input Stream for the header.
+ final BufferedInputStream bufferedIn = new BufferedInputStream(in);
+ bufferedIn.mark(1024 * 1024);
+ final RecordSchema schema = getSchema(flowFile, new NonCloseableInputStream(bufferedIn));
+ bufferedIn.reset();
+
+ return new CSVRecordReader(bufferedIn, logger, schema, csvFormat, dateFormat, timeFormat, timestampFormat);
+ }
+
+ @Override
+ protected SchemaAccessStrategy getSchemaAccessStrategy(final String allowableValue, final SchemaRegistry schemaRegistry) {
+ if (allowableValue.equalsIgnoreCase(headerDerivedAllowableValue.getValue())) {
+ return headerDerivedSchemaStrategy;
+ }
+
+ return super.getSchemaAccessStrategy(allowableValue, schemaRegistry);
}
+ @Override
+ protected List<AllowableValue> getSchemaAccessStrategyValues() {
+ final List<AllowableValue> allowableValues = new ArrayList<>(super.getSchemaAccessStrategyValues());
+ allowableValues.add(headerDerivedAllowableValue);
+ return allowableValues;
+ }
+
+ @Override
+ protected AllowableValue getDefaultSchemaAccessStrategy() {
+ return headerDerivedAllowableValue;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
index d02768c..241d604 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
@@ -34,6 +34,7 @@ import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.util.DataTypeUtils;
@@ -48,13 +49,14 @@ public class CSVRecordReader implements RecordReader {
public CSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat,
final String dateFormat, final String timeFormat, final String timestampFormat) throws IOException {
- final Reader reader = new InputStreamReader(new BOMInputStream(in));
- csvParser = new CSVParser(reader, csvFormat);
-
this.schema = schema;
this.dateFormat = dateFormat;
this.timeFormat = timeFormat;
this.timestampFormat = timestampFormat;
+
+ final Reader reader = new InputStreamReader(new BOMInputStream(in));
+ final CSVFormat withHeader = csvFormat.withHeader(schema.getFieldNames().toArray(new String[0]));
+ csvParser = new CSVParser(reader, withHeader);
}
@Override
@@ -64,15 +66,27 @@ public class CSVRecordReader implements RecordReader {
for (final CSVRecord csvRecord : csvParser) {
final Map<String, Object> rowValues = new HashMap<>(schema.getFieldCount());
- for (final String fieldName : schema.getFieldNames()) {
- final String rawValue = csvRecord.get(fieldName);
+ for (final RecordField recordField : schema.getFields()) {
+ String rawValue = csvRecord.get(recordField.getFieldName());
+ if (rawValue == null) {
+ for (final String alias : recordField.getAliases()) {
+ rawValue = csvRecord.get(alias);
+ if (rawValue != null) {
+ break;
+ }
+ }
+ }
+
+ final String fieldName = recordField.getFieldName();
if (rawValue == null) {
rowValues.put(fieldName, null);
continue;
}
- final Object converted = convert(rawValue, schema.getDataType(fieldName).orElse(null));
- rowValues.put(fieldName, converted);
+ final Object converted = convert(rawValue, recordField.getDataType(), fieldName);
+ if (converted != null) {
+ rowValues.put(fieldName, converted);
+ }
}
return new MapRecord(schema, rowValues);
@@ -86,7 +100,7 @@ public class CSVRecordReader implements RecordReader {
return schema;
}
- protected Object convert(final String value, final DataType dataType) {
+ protected Object convert(final String value, final DataType dataType, final String fieldName) {
if (dataType == null || value == null) {
return value;
}
@@ -97,7 +111,7 @@ public class CSVRecordReader implements RecordReader {
return null;
}
- return DataTypeUtils.convertType(trimmed, dataType, dateFormat, timeFormat, timestampFormat);
+ return DataTypeUtils.convertType(trimmed, dataType, dateFormat, timeFormat, timestampFormat, fieldName);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
index 6a7b758..95c86e7 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
@@ -17,6 +17,8 @@
package org.apache.nifi.csv;
+import java.io.IOException;
+import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
@@ -26,10 +28,13 @@ import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.serialization.DateTimeTextRecordSetWriter;
import org.apache.nifi.serialization.RecordSetWriter;
import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.record.RecordSchema;
@Tags({"csv", "result", "set", "recordset", "record", "writer", "serializer", "row", "tsv", "tab", "separated", "delimited"})
@CapabilityDescription("Writes the contents of a RecordSet as CSV data. The first line written "
@@ -37,12 +42,14 @@ import org.apache.nifi.serialization.RecordSetWriterFactory;
public class CSVRecordSetWriter extends DateTimeTextRecordSetWriter implements RecordSetWriterFactory {
private volatile CSVFormat csvFormat;
+ private volatile boolean includeHeader;
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
properties.add(CSVUtils.CSV_FORMAT);
properties.add(CSVUtils.VALUE_SEPARATOR);
+ properties.add(CSVUtils.INCLUDE_HEADER_LINE);
properties.add(CSVUtils.QUOTE_CHAR);
properties.add(CSVUtils.ESCAPE_CHAR);
properties.add(CSVUtils.COMMENT_MARKER);
@@ -57,11 +64,12 @@ public class CSVRecordSetWriter extends DateTimeTextRecordSetWriter implements R
@OnEnabled
public void storeCsvFormat(final ConfigurationContext context) {
this.csvFormat = CSVUtils.createCSVFormat(context);
+ this.includeHeader = context.getProperty(CSVUtils.INCLUDE_HEADER_LINE).asBoolean();
}
@Override
- public RecordSetWriter createWriter(final ComponentLog logger) {
- return new WriteCSVResult(csvFormat, getDateFormat(), getTimeFormat(), getTimestampFormat());
+ public RecordSetWriter createWriter(final ComponentLog logger, final FlowFile flowFile, final InputStream in) throws SchemaNotFoundException, IOException {
+ final RecordSchema schema = getSchema(flowFile, in);
+ return new WriteCSVResult(csvFormat, schema, getSchemaAccessWriter(schema), getDateFormat(), getTimeFormat(), getTimestampFormat(), includeHeader);
}
-
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVUtils.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVUtils.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVUtils.java
index e23b6e1..1048d21 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVUtils.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVUtils.java
@@ -61,6 +61,18 @@ public class CSVUtils {
.defaultValue("\"")
.required(true)
.build();
+ static final PropertyDescriptor SKIP_HEADER_LINE = new PropertyDescriptor.Builder()
+ .name("Skip Header Line")
+ .description("Specifies whether or not the first line of CSV should be considered a Header and skipped. If the Schema Access Strategy "
+ + "indicates that the columns must be defined in the header, then this property will be ignored, since the header must always be "
+ + "present and won't be processed as a Record. Otherwise, this property should be 'true' if the first non-comment line of CSV "
+ + "contains header information that needs to be ignored.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
static final PropertyDescriptor COMMENT_MARKER = new PropertyDescriptor.Builder()
.name("Comment Marker")
.description("The character that is used to denote the start of a comment. Any line that begins with this comment will be ignored.")
@@ -124,7 +136,13 @@ public class CSVUtils {
.defaultValue("\\n")
.required(true)
.build();
-
+ static final PropertyDescriptor INCLUDE_HEADER_LINE = new PropertyDescriptor.Builder()
+ .name("Include Header Line")
+ .description("Specifies whether or not the CSV column names should be written out as the first line.")
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .required(true)
+ .build();
static CSVFormat createCSVFormat(final ConfigurationContext context) {
final String formatName = context.getProperty(CSV_FORMAT).getValue();
@@ -156,8 +174,11 @@ public class CSVUtils {
final char valueSeparator = getChar(context, VALUE_SEPARATOR);
CSVFormat format = CSVFormat.newFormat(valueSeparator)
.withAllowMissingColumnNames()
- .withIgnoreEmptyLines()
- .withFirstRecordAsHeader();
+ .withIgnoreEmptyLines();
+
+ if (context.getProperty(SKIP_HEADER_LINE).asBoolean()) {
+ format = format.withFirstRecordAsHeader();
+ }
format = format.withQuote(getChar(context, QUOTE_CHAR));
format = format.withEscape(getChar(context, ESCAPE_CHAR));
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/WriteCSVResult.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/WriteCSVResult.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/WriteCSVResult.java
index e0eb813..7c53ace 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/WriteCSVResult.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/WriteCSVResult.java
@@ -21,38 +21,41 @@ import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.util.Collections;
-import java.util.Optional;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVPrinter;
+import org.apache.nifi.schema.access.SchemaAccessWriter;
import org.apache.nifi.serialization.RecordSetWriter;
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.RecordField;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.RecordSet;
import org.apache.nifi.stream.io.NonCloseableOutputStream;
public class WriteCSVResult implements RecordSetWriter {
private final CSVFormat csvFormat;
+ private final RecordSchema recordSchema;
+ private final SchemaAccessWriter schemaWriter;
private final String dateFormat;
private final String timeFormat;
private final String timestampFormat;
+ private final boolean includeHeaderLine;
- public WriteCSVResult(final CSVFormat csvFormat, final String dateFormat, final String timeFormat, final String timestampFormat) {
+ public WriteCSVResult(final CSVFormat csvFormat, final RecordSchema recordSchema, final SchemaAccessWriter schemaWriter,
+ final String dateFormat, final String timeFormat, final String timestampFormat, final boolean includeHeaderLine) {
this.csvFormat = csvFormat;
+ this.recordSchema = recordSchema;
+ this.schemaWriter = schemaWriter;
this.dateFormat = dateFormat;
this.timeFormat = timeFormat;
this.timestampFormat = timestampFormat;
+ this.includeHeaderLine = includeHeaderLine;
}
- private String getFormat(final Record record, final String fieldName) {
- final Optional<DataType> dataTypeOption = record.getSchema().getDataType(fieldName);
- if (!dataTypeOption.isPresent()) {
- return null;
- }
-
- final DataType dataType = dataTypeOption.get();
+ private String getFormat(final Record record, final RecordField field) {
+ final DataType dataType = field.getDataType();
switch (dataType.getFieldType()) {
case DATE:
return dateFormat == null ? dataType.getFormat() : dateFormat;
@@ -69,9 +72,10 @@ public class WriteCSVResult implements RecordSetWriter {
public WriteResult write(final RecordSet rs, final OutputStream rawOut) throws IOException {
int count = 0;
- final RecordSchema schema = rs.getSchema();
- final String[] columnNames = schema.getFieldNames().toArray(new String[0]);
- final CSVFormat formatWithHeader = csvFormat.withHeader(columnNames);
+ final String[] columnNames = recordSchema.getFieldNames().toArray(new String[0]);
+ final CSVFormat formatWithHeader = csvFormat.withHeader(columnNames).withSkipHeaderRecord(!includeHeaderLine);
+
+ schemaWriter.writeHeader(recordSchema, rawOut);
try (final OutputStream nonCloseable = new NonCloseableOutputStream(rawOut);
final OutputStreamWriter streamWriter = new OutputStreamWriter(nonCloseable);
@@ -80,10 +84,10 @@ public class WriteCSVResult implements RecordSetWriter {
try {
Record record;
while ((record = rs.next()) != null) {
- final Object[] colVals = new Object[schema.getFieldCount()];
+ final Object[] colVals = new Object[recordSchema.getFieldCount()];
int i = 0;
- for (final String fieldName : schema.getFieldNames()) {
- colVals[i++] = record.getAsString(fieldName, getFormat(record, fieldName));
+ for (final RecordField recordField : recordSchema.getFields()) {
+ colVals[i++] = record.getAsString(recordField, getFormat(record, recordField));
}
printer.printRecord(colVals);
@@ -94,7 +98,7 @@ public class WriteCSVResult implements RecordSetWriter {
}
}
- return WriteResult.of(count, Collections.emptyMap());
+ return WriteResult.of(count, schemaWriter.getAttributes(recordSchema));
}
@Override
@@ -108,8 +112,8 @@ public class WriteCSVResult implements RecordSetWriter {
final RecordSchema schema = record.getSchema();
final Object[] colVals = new Object[schema.getFieldCount()];
int i = 0;
- for (final String fieldName : schema.getFieldNames()) {
- colVals[i++] = record.getAsString(fieldName, getFormat(record, fieldName));
+ for (final RecordField recordField : schema.getFields()) {
+ colVals[i++] = record.getAsString(recordField, getFormat(record, recordField));
}
printer.printRecord(colVals);
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokReader.java
index f444b8a..778c738 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokReader.java
@@ -22,39 +22,62 @@ import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.schema.access.SchemaAccessStrategy;
+import org.apache.nifi.schema.access.SchemaField;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.schemaregistry.services.SchemaRegistry;
import org.apache.nifi.serialization.RecordReader;
-import org.apache.nifi.serialization.RowRecordReaderFactory;
-import org.apache.nifi.serialization.SchemaRegistryRecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.SchemaRegistryService;
+import org.apache.nifi.serialization.SimpleRecordSchema;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import io.thekraken.grok.api.Grok;
+import io.thekraken.grok.api.GrokUtils;
import io.thekraken.grok.api.exception.GrokException;
@Tags({"grok", "logs", "logfiles", "parse", "unstructured", "text", "record", "reader", "regex", "pattern", "logstash"})
@CapabilityDescription("Provides a mechanism for reading unstructured text data, such as log files, and structuring the data "
+ "so that it can be processed. The service is configured using Grok patterns. "
+ "The service reads from a stream of data and splits each message that it finds into a separate Record, each containing the fields that are configured. "
- + "If a line in the input does not match the expected message pattern, the line of text is considered to be part of the previous "
- + "message, with the exception of stack traces. A stack trace that is found at the end of a log message is considered to be part "
- + "of the previous message but is added to the 'STACK_TRACE' field of the Record. If a record has no stack trace, it will have a NULL value "
- + "for the STACK_TRACE field. All fields that are parsed are considered to be of type String by default. If there is need to change the type of a field, "
- + "this can be accomplished by configuring the Schema Registry to use and adding the appropriate schema.")
-public class GrokReader extends SchemaRegistryRecordReader implements RowRecordReaderFactory {
+ + "If a line in the input does not match the expected message pattern, the line of text is either considered to be part of the previous "
+ + "message or is skipped, depending on the configuration, with the exception of stack traces. A stack trace that is found at the end of "
+ + "a log message is considered to be part of the previous message but is added to the 'stackTrace' field of the Record. If a record has "
+ + "no stack trace, it will have a NULL value for the stackTrace field (assuming that the schema does in fact include a stackTrace field of type String).")
+public class GrokReader extends SchemaRegistryService implements RecordReaderFactory {
private volatile Grok grok;
- private volatile boolean useSchemaRegistry;
+ private volatile boolean appendUnmatchedLine;
+ private volatile RecordSchema recordSchema;
private static final String DEFAULT_PATTERN_NAME = "/default-grok-patterns.txt";
+ static final AllowableValue APPEND_TO_PREVIOUS_MESSAGE = new AllowableValue("append-to-previous-message", "Append to Previous Message",
+ "The line of text that does not match the Grok Expression will be appended to the last field of the prior message.");
+ static final AllowableValue SKIP_LINE = new AllowableValue("skip-line", "Skip Line",
+ "The line of text that does not match the Grok Expression will be skipped.");
+
+ static final AllowableValue STRING_FIELDS_FROM_GROK_EXPRESSION = new AllowableValue("string-fields-from-grok-expression", "Use String Fields From Grok Expression",
+ "The schema will be derived by using the field names present in the Grok Expression. All fields will be assumed to be of type String. Additionally, a field will be included "
+ + "with a name of 'stackTrace' and a type of String.");
+
static final PropertyDescriptor PATTERN_FILE = new PropertyDescriptor.Builder()
.name("Grok Pattern File")
.description("Path to a file that contains Grok Patterns to use for parsing logs. If not specified, a built-in default Pattern file "
@@ -73,11 +96,22 @@ public class GrokReader extends SchemaRegistryRecordReader implements RowRecordR
.required(true)
.build();
+ static final PropertyDescriptor NO_MATCH_BEHAVIOR = new PropertyDescriptor.Builder()
+ .name("no-match-behavior")
+ .displayName("No Match Behavior")
+ .description("If a line of text is encountered and it does not match the given Grok Expression, and it is not part of a stack trace, "
+ + "this property specifies how the text should be processed.")
+ .allowableValues(APPEND_TO_PREVIOUS_MESSAGE, SKIP_LINE)
+ .defaultValue(APPEND_TO_PREVIOUS_MESSAGE.getValue())
+ .required(true)
+ .build();
+
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
properties.add(PATTERN_FILE);
properties.add(GROK_EXPRESSION);
+ properties.add(NO_MATCH_BEHAVIOR);
return properties;
}
@@ -95,17 +129,79 @@ public class GrokReader extends SchemaRegistryRecordReader implements RowRecordR
}
grok.compile(context.getProperty(GROK_EXPRESSION).getValue());
- useSchemaRegistry = context.getProperty(OPTIONAL_SCHEMA_NAME).isSet() && context.getProperty(OPTIONAL_SCHEMA_REGISTRY).isSet();
+
+ appendUnmatchedLine = context.getProperty(NO_MATCH_BEHAVIOR).getValue().equalsIgnoreCase(APPEND_TO_PREVIOUS_MESSAGE.getValue());
+
+ this.recordSchema = createRecordSchema(grok);
+ }
+
+ static RecordSchema createRecordSchema(final Grok grok) {
+ final List<RecordField> fields = new ArrayList<>();
+
+ String grokExpression = grok.getOriginalGrokPattern();
+ while (grokExpression.length() > 0) {
+ final Matcher matcher = GrokUtils.GROK_PATTERN.matcher(grokExpression);
+ if (matcher.find()) {
+ final Map<String, String> namedGroups = GrokUtils.namedGroups(matcher, grokExpression);
+ final String fieldName = namedGroups.get("subname");
+
+ DataType dataType = RecordFieldType.STRING.getDataType();
+ final RecordField recordField = new RecordField(fieldName, dataType);
+ fields.add(recordField);
+
+ if (grokExpression.length() > matcher.end() + 1) {
+ grokExpression = grokExpression.substring(matcher.end() + 1);
+ } else {
+ break;
+ }
+ }
+ }
+
+ fields.add(new RecordField(GrokRecordReader.STACK_TRACE_COLUMN_NAME, RecordFieldType.STRING.getDataType()));
+
+ final RecordSchema schema = new SimpleRecordSchema(fields);
+ return schema;
}
+
@Override
- protected boolean isSchemaRequired() {
- return false;
+ protected List<AllowableValue> getSchemaAccessStrategyValues() {
+ final List<AllowableValue> allowableValues = new ArrayList<>();
+ allowableValues.add(STRING_FIELDS_FROM_GROK_EXPRESSION);
+ allowableValues.addAll(super.getSchemaAccessStrategyValues());
+ return allowableValues;
}
@Override
- public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException {
- final RecordSchema schema = useSchemaRegistry ? getSchema(flowFile) : null;
- return new GrokRecordReader(in, grok, schema);
+ protected AllowableValue getDefaultSchemaAccessStrategy() {
+ return STRING_FIELDS_FROM_GROK_EXPRESSION;
+ }
+
+ @Override
+ protected SchemaAccessStrategy getSchemaAccessStrategy(final String allowableValue, final SchemaRegistry schemaRegistry) {
+ if (allowableValue.equalsIgnoreCase(STRING_FIELDS_FROM_GROK_EXPRESSION.getValue())) {
+ return new SchemaAccessStrategy() {
+ private final Set<SchemaField> schemaFields = EnumSet.noneOf(SchemaField.class);
+
+ @Override
+ public RecordSchema getSchema(final FlowFile flowFile, final InputStream contentStream, final ConfigurationContext context) throws SchemaNotFoundException {
+ return recordSchema;
+ }
+
+ @Override
+ public Set<SchemaField> getSuppliedSchemaFields() {
+ return schemaFields;
+ }
+ };
+ } else {
+ return super.getSchemaAccessStrategy(allowableValue, schemaRegistry);
+ }
+ }
+
+
+ @Override
+ public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException, SchemaNotFoundException {
+ final RecordSchema schema = getSchema(flowFile, in);
+ return new GrokRecordReader(in, grok, schema, appendUnmatchedLine);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokRecordReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokRecordReader.java
index 458dbd8..5859f6f 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokRecordReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/grok/GrokRecordReader.java
@@ -21,40 +21,34 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
-import java.text.ParseException;
-import java.util.ArrayList;
import java.util.Collections;
-import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.TimeZone;
-import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
-import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.MapRecord;
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.util.DataTypeUtils;
import io.thekraken.grok.api.Grok;
-import io.thekraken.grok.api.GrokUtils;
import io.thekraken.grok.api.Match;
public class GrokRecordReader implements RecordReader {
private final BufferedReader reader;
private final Grok grok;
+ private final boolean append;
private RecordSchema schema;
private String nextLine;
- static final String STACK_TRACE_COLUMN_NAME = "STACK_TRACE";
+ static final String STACK_TRACE_COLUMN_NAME = "stackTrace";
private static final Pattern STACK_TRACE_PATTERN = Pattern.compile(
"^\\s*(?:(?: |\\t)+at )|"
+ "(?:(?: |\\t)+\\[CIRCULAR REFERENCE\\:)|"
@@ -62,21 +56,11 @@ public class GrokRecordReader implements RecordReader {
+ "(?:Suppressed\\: )|"
+ "(?:\\s+... \\d+ (?:more|common frames? omitted)$)");
- private static final FastDateFormat TIME_FORMAT_DATE;
- private static final FastDateFormat TIME_FORMAT_TIME;
- private static final FastDateFormat TIME_FORMAT_TIMESTAMP;
-
- static {
- final TimeZone gmt = TimeZone.getTimeZone("GMT");
- TIME_FORMAT_DATE = FastDateFormat.getInstance("yyyy-MM-dd", gmt);
- TIME_FORMAT_TIME = FastDateFormat.getInstance("HH:mm:ss", gmt);
- TIME_FORMAT_TIMESTAMP = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss", gmt);
- }
-
- public GrokRecordReader(final InputStream in, final Grok grok, final RecordSchema schema) {
+ public GrokRecordReader(final InputStream in, final Grok grok, final RecordSchema schema, final boolean append) {
this.reader = new BufferedReader(new InputStreamReader(in));
this.grok = grok;
this.schema = schema;
+ this.append = append;
}
@Override
@@ -115,7 +99,7 @@ public class GrokRecordReader implements RecordReader {
if (isStartOfStackTrace(nextLine)) {
stackTrace = readStackTrace(nextLine);
break;
- } else {
+ } else if (append) {
toAppend.append("\n").append(nextLine);
}
} else {
@@ -128,20 +112,34 @@ public class GrokRecordReader implements RecordReader {
final List<DataType> fieldTypes = schema.getDataTypes();
final Map<String, Object> values = new HashMap<>(fieldTypes.size());
- for (final String fieldName : schema.getFieldNames()) {
- final Object value = valueMap.get(fieldName);
+ for (final RecordField field : schema.getFields()) {
+ Object value = valueMap.get(field.getFieldName());
+ if (value == null) {
+ for (final String alias : field.getAliases()) {
+ value = valueMap.get(alias);
+ if (value != null) {
+ break;
+ }
+ }
+ }
+
+ final String fieldName = field.getFieldName();
if (value == null) {
values.put(fieldName, null);
continue;
}
- final DataType fieldType = schema.getDataType(fieldName).orElse(null);
- final Object converted = convert(fieldType, value.toString());
+ final DataType fieldType = field.getDataType();
+ final Object converted = convert(fieldType, value.toString(), fieldName);
values.put(fieldName, converted);
}
- final String lastFieldBeforeStackTrace = schema.getFieldNames().get(schema.getFieldCount() - 2);
- if (toAppend.length() > 0) {
+ if (append && toAppend.length() > 0) {
+ final String lastFieldName = schema.getField(schema.getFieldCount() - 1).getFieldName();
+
+ final int fieldIndex = STACK_TRACE_COLUMN_NAME.equals(lastFieldName) ? schema.getFieldCount() - 2 : schema.getFieldCount() - 1;
+ final String lastFieldBeforeStackTrace = schema.getFieldNames().get(fieldIndex);
+
final Object existingValue = values.get(lastFieldBeforeStackTrace);
final String updatedValue = existingValue == null ? toAppend.toString() : existingValue + toAppend.toString();
values.put(lastFieldBeforeStackTrace, updatedValue);
@@ -205,7 +203,7 @@ public class GrokRecordReader implements RecordReader {
}
- protected Object convert(final DataType fieldType, final String string) {
+ protected Object convert(final DataType fieldType, final String string, final String fieldName) {
if (fieldType == null) {
return string;
}
@@ -220,79 +218,12 @@ public class GrokRecordReader implements RecordReader {
return null;
}
- switch (fieldType.getFieldType()) {
- case BOOLEAN:
- return Boolean.parseBoolean(string);
- case BYTE:
- return Byte.parseByte(string);
- case SHORT:
- return Short.parseShort(string);
- case INT:
- return Integer.parseInt(string);
- case LONG:
- return Long.parseLong(string);
- case FLOAT:
- return Float.parseFloat(string);
- case DOUBLE:
- return Double.parseDouble(string);
- case DATE:
- try {
- Date date = TIME_FORMAT_DATE.parse(string);
- return new java.sql.Date(date.getTime());
- } catch (ParseException e) {
- return null;
- }
- case TIME:
- try {
- Date date = TIME_FORMAT_TIME.parse(string);
- return new java.sql.Time(date.getTime());
- } catch (ParseException e) {
- return null;
- }
- case TIMESTAMP:
- try {
- Date date = TIME_FORMAT_TIMESTAMP.parse(string);
- return new java.sql.Timestamp(date.getTime());
- } catch (ParseException e) {
- return null;
- }
- case STRING:
- default:
- return string;
- }
+ return DataTypeUtils.convertType(string, fieldType, fieldName);
}
@Override
public RecordSchema getSchema() {
- if (schema != null) {
- return schema;
- }
-
- final List<RecordField> fields = new ArrayList<>();
-
- String grokExpression = grok.getOriginalGrokPattern();
- while (grokExpression.length() > 0) {
- final Matcher matcher = GrokUtils.GROK_PATTERN.matcher(grokExpression);
- if (matcher.find()) {
- final Map<String, String> namedGroups = GrokUtils.namedGroups(matcher, grokExpression);
- final String fieldName = namedGroups.get("subname");
-
- DataType dataType = RecordFieldType.STRING.getDataType();
- final RecordField recordField = new RecordField(fieldName, dataType);
- fields.add(recordField);
-
- if (grokExpression.length() > matcher.end() + 1) {
- grokExpression = grokExpression.substring(matcher.end() + 1);
- } else {
- break;
- }
- }
- }
-
- fields.add(new RecordField(STACK_TRACE_COLUMN_NAME, RecordFieldType.STRING.getDataType()));
-
- schema = new SimpleRecordSchema(fields);
return schema;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
index ad04912..c5c5fb0 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
@@ -19,20 +19,12 @@ package org.apache.nifi.json;
import java.io.IOException;
import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
import java.util.Optional;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
-import org.apache.nifi.serialization.SimpleRecordSchema;
-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.codehaus.jackson.JsonFactory;
import org.codehaus.jackson.JsonNode;
@@ -98,55 +90,6 @@ public abstract class AbstractJsonRowRecordReader implements RecordReader {
}
}
- protected DataType determineFieldType(final JsonNode node) {
- if (node.isDouble()) {
- return RecordFieldType.DOUBLE.getDataType();
- }
- if (node.isBoolean()) {
- return RecordFieldType.BOOLEAN.getDataType();
- }
- if (node.isFloatingPointNumber()) {
- return RecordFieldType.FLOAT.getDataType();
- }
- if (node.isBigInteger()) {
- return RecordFieldType.BIGINT.getDataType();
- }
- if (node.isBigDecimal()) {
- return RecordFieldType.DOUBLE.getDataType();
- }
- if (node.isLong()) {
- return RecordFieldType.LONG.getDataType();
- }
- if (node.isInt()) {
- return RecordFieldType.INT.getDataType();
- }
- if (node.isTextual()) {
- return RecordFieldType.STRING.getDataType();
- }
- if (node.isArray()) {
- return RecordFieldType.ARRAY.getDataType();
- }
-
- final RecordSchema childSchema = determineSchema(node);
- return RecordFieldType.RECORD.getRecordDataType(childSchema);
- }
-
- protected RecordSchema determineSchema(final JsonNode jsonNode) {
- final List<RecordField> recordFields = new ArrayList<>();
-
- final Iterator<Map.Entry<String, JsonNode>> itr = jsonNode.getFields();
- while (itr.hasNext()) {
- final Map.Entry<String, JsonNode> entry = itr.next();
- final String elementName = entry.getKey();
- final JsonNode node = entry.getValue();
-
- DataType dataType = determineFieldType(node);
- recordFields.add(new RecordField(elementName, dataType));
- }
-
- return new SimpleRecordSchema(recordFields);
- }
-
protected Object getRawNodeValue(final JsonNode fieldNode) throws IOException {
if (fieldNode == null || !fieldNode.isValueNode()) {
return null;
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathReader.java
index 467ecf8..2d11a9b 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathReader.java
@@ -36,11 +36,12 @@ import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.serialization.DateTimeUtils;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
-import org.apache.nifi.serialization.RowRecordReaderFactory;
-import org.apache.nifi.serialization.SchemaRegistryRecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.SchemaRegistryService;
import org.apache.nifi.serialization.record.RecordSchema;
import com.jayway.jsonpath.JsonPath;
@@ -56,7 +57,7 @@ import com.jayway.jsonpath.JsonPath;
+ "field whose name is the same as the property name.",
description="User-defined properties identifiy how to extract specific fields from a JSON object in order to create a Record",
supportsExpressionLanguage=false)
-public class JsonPathReader extends SchemaRegistryRecordReader implements RowRecordReaderFactory {
+public class JsonPathReader extends SchemaRegistryService implements RecordReaderFactory {
private volatile String dateFormat;
private volatile String timeFormat;
@@ -127,8 +128,8 @@ public class JsonPathReader extends SchemaRegistryRecordReader implements RowRec
}
@Override
- public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException, MalformedRecordException {
- final RecordSchema schema = getSchema(flowFile);
+ public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException, MalformedRecordException, SchemaNotFoundException {
+ final RecordSchema schema = getSchema(flowFile, in);
return new JsonPathRowRecordReader(jsonPaths, schema, in, logger, dateFormat, timeFormat, timestampFormat);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
index a0f3c32..8675e0e 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
@@ -30,6 +30,7 @@ import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.MapRecord;
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;
@@ -106,7 +107,10 @@ public class JsonPathRowRecordReader extends AbstractJsonRowRecordReader {
value = null;
}
- value = convert(value, desiredType);
+ final Optional<RecordField> field = schema.getField(fieldName);
+ final Object defaultValue = field.isPresent() ? field.get().getDefaultValue() : null;
+
+ value = convert(value, desiredType, fieldName, defaultValue);
values.put(fieldName, value);
}
@@ -115,9 +119,9 @@ public class JsonPathRowRecordReader extends AbstractJsonRowRecordReader {
@SuppressWarnings("unchecked")
- protected Object convert(final Object value, final DataType dataType) {
+ protected Object convert(final Object value, final DataType dataType, final String fieldName, final Object defaultValue) {
if (value == null) {
- return null;
+ return defaultValue;
}
if (value instanceof List) {
@@ -131,7 +135,7 @@ public class JsonPathRowRecordReader extends AbstractJsonRowRecordReader {
final Object[] coercedValues = new Object[list.size()];
int i = 0;
for (final Object rawValue : list) {
- coercedValues[i++] = DataTypeUtils.convertType(rawValue, arrayType.getElementType(), dateFormat, timeFormat, timestampFormat);
+ coercedValues[i++] = convert(rawValue, arrayType.getElementType(), fieldName, null);
}
return coercedValues;
}
@@ -147,14 +151,17 @@ public class JsonPathRowRecordReader extends AbstractJsonRowRecordReader {
final String key = entry.getKey();
final Optional<DataType> desiredTypeOption = childSchema.getDataType(key);
if (desiredTypeOption.isPresent()) {
- final Object coercedValue = DataTypeUtils.convertType(entry.getValue(), desiredTypeOption.get(), dateFormat, timeFormat, timestampFormat);
+ final Optional<RecordField> field = childSchema.getField(key);
+ final Object defaultFieldValue = field.isPresent() ? field.get().getDefaultValue() : null;
+
+ final Object coercedValue = convert(entry.getValue(), desiredTypeOption.get(), fieldName + "." + key, defaultFieldValue);
coercedValues.put(key, coercedValue);
}
}
return new MapRecord(childSchema, coercedValues);
} else {
- return DataTypeUtils.convertType(value, dataType, dateFormat, timeFormat, timestampFormat);
+ return DataTypeUtils.convertType(value, dataType, dateFormat, timeFormat, timestampFormat, fieldName);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
index d09f135..e6b5c02 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
@@ -17,6 +17,8 @@
package org.apache.nifi.json;
+import java.io.IOException;
+import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
@@ -25,10 +27,13 @@ import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.serialization.DateTimeTextRecordSetWriter;
import org.apache.nifi.serialization.RecordSetWriter;
import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.record.RecordSchema;
@Tags({"json", "resultset", "writer", "serialize", "record", "recordset", "row"})
@CapabilityDescription("Writes the results of a RecordSet as a JSON Array. Even if the RecordSet "
@@ -59,8 +64,9 @@ public class JsonRecordSetWriter extends DateTimeTextRecordSetWriter implements
}
@Override
- public RecordSetWriter createWriter(final ComponentLog logger) {
- return new WriteJsonResult(logger, prettyPrint, getDateFormat(), getTimeFormat(), getTimestampFormat());
+ public RecordSetWriter createWriter(final ComponentLog logger, final FlowFile flowFile, final InputStream flowFileContent) throws SchemaNotFoundException, IOException {
+ final RecordSchema schema = getSchema(flowFile, flowFileContent);
+ return new WriteJsonResult(logger, schema, getSchemaAccessWriter(schema), prettyPrint, getDateFormat(), getTimeFormat(), getTimestampFormat());
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeReader.java
index 1abb1f4..1dd9834 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeReader.java
@@ -30,11 +30,12 @@ import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.serialization.DateTimeUtils;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
-import org.apache.nifi.serialization.RowRecordReaderFactory;
-import org.apache.nifi.serialization.SchemaRegistryRecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.SchemaRegistryService;
@Tags({"json", "tree", "record", "reader", "parser"})
@CapabilityDescription("Parses JSON into individual Record objects. The Record that is produced will contain all top-level "
@@ -45,7 +46,7 @@ import org.apache.nifi.serialization.SchemaRegistryRecordReader;
+ "a field that is not present in the schema, that field will be skipped. "
+ "See the Usage of the Controller Service for more information and examples.")
@SeeAlso(JsonPathReader.class)
-public class JsonTreeReader extends SchemaRegistryRecordReader implements RowRecordReaderFactory {
+public class JsonTreeReader extends SchemaRegistryService implements RecordReaderFactory {
private volatile String dateFormat;
private volatile String timeFormat;
@@ -68,7 +69,7 @@ public class JsonTreeReader extends SchemaRegistryRecordReader implements RowRec
}
@Override
- public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException, MalformedRecordException {
- return new JsonTreeRowRecordReader(in, logger, getSchema(flowFile), dateFormat, timeFormat, timestampFormat);
+ public RecordReader createRecordReader(final FlowFile flowFile, final InputStream in, final ComponentLog logger) throws IOException, MalformedRecordException, SchemaNotFoundException {
+ return new JsonTreeRowRecordReader(in, logger, getSchema(flowFile, in), dateFormat, timeFormat, timestampFormat);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
index c8d07f4..301b724 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
@@ -19,17 +19,23 @@ package org.apache.nifi.json;
import java.io.IOException;
import java.io.InputStream;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.MapRecord;
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.MapDataType;
import org.apache.nifi.serialization.record.type.RecordDataType;
import org.apache.nifi.serialization.record.util.DataTypeUtils;
import org.codehaus.jackson.JsonNode;
@@ -55,6 +61,10 @@ public class JsonTreeRowRecordReader extends AbstractJsonRowRecordReader {
@Override
protected Record convertJsonNodeToRecord(final JsonNode jsonNode, final RecordSchema schema) throws IOException, MalformedRecordException {
+ return convertJsonNodeToRecord(jsonNode, schema, "");
+ }
+
+ private Record convertJsonNodeToRecord(final JsonNode jsonNode, final RecordSchema schema, final String fieldNamePrefix) throws IOException, MalformedRecordException {
if (jsonNode == null) {
return null;
}
@@ -63,10 +73,19 @@ public class JsonTreeRowRecordReader extends AbstractJsonRowRecordReader {
for (int i = 0; i < schema.getFieldCount(); i++) {
final RecordField field = schema.getField(i);
final String fieldName = field.getFieldName();
- final JsonNode fieldNode = jsonNode.get(fieldName);
+
+ JsonNode fieldNode = jsonNode.get(fieldName);
+ if (fieldNode == null) {
+ for (final String alias : field.getAliases()) {
+ fieldNode = jsonNode.get(alias);
+ if (fieldNode != null) {
+ break;
+ }
+ }
+ }
final DataType desiredType = field.getDataType();
- final Object value = convertField(fieldNode, fieldName, desiredType);
+ final Object value = convertField(fieldNode, fieldNamePrefix + fieldName, desiredType);
values.put(fieldName, value);
}
@@ -80,42 +99,50 @@ public class JsonTreeRowRecordReader extends AbstractJsonRowRecordReader {
switch (desiredType.getFieldType()) {
case BOOLEAN:
- return DataTypeUtils.toBoolean(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toBoolean(getRawNodeValue(fieldNode), fieldName);
case BYTE:
- return DataTypeUtils.toByte(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toByte(getRawNodeValue(fieldNode), fieldName);
case CHAR:
- return DataTypeUtils.toCharacter(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toCharacter(getRawNodeValue(fieldNode), fieldName);
case DOUBLE:
- return DataTypeUtils.toDouble(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toDouble(getRawNodeValue(fieldNode), fieldName);
case FLOAT:
- return DataTypeUtils.toFloat(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toFloat(getRawNodeValue(fieldNode), fieldName);
case INT:
- return DataTypeUtils.toInteger(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toInteger(getRawNodeValue(fieldNode), fieldName);
case LONG:
- return DataTypeUtils.toLong(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toLong(getRawNodeValue(fieldNode), fieldName);
case SHORT:
- return DataTypeUtils.toShort(getRawNodeValue(fieldNode));
+ return DataTypeUtils.toShort(getRawNodeValue(fieldNode), fieldName);
case STRING:
return DataTypeUtils.toString(getRawNodeValue(fieldNode), dateFormat, timeFormat, timestampFormat);
case DATE:
- return DataTypeUtils.toDate(getRawNodeValue(fieldNode), dateFormat);
+ return DataTypeUtils.toDate(getRawNodeValue(fieldNode), dateFormat, fieldName);
case TIME:
- return DataTypeUtils.toTime(getRawNodeValue(fieldNode), timeFormat);
+ return DataTypeUtils.toTime(getRawNodeValue(fieldNode), timeFormat, fieldName);
case TIMESTAMP:
- return DataTypeUtils.toTimestamp(getRawNodeValue(fieldNode), timestampFormat);
+ return DataTypeUtils.toTimestamp(getRawNodeValue(fieldNode), timestampFormat, fieldName);
+ case MAP: {
+ final DataType valueType = ((MapDataType) desiredType).getValueType();
+
+ final Map<String, Object> map = new HashMap<>();
+ final Iterator<String> fieldNameItr = fieldNode.getFieldNames();
+ while (fieldNameItr.hasNext()) {
+ final String childName = fieldNameItr.next();
+ final JsonNode childNode = fieldNode.get(childName);
+ final Object childValue = convertField(childNode, fieldName + "." + childName, valueType);
+ map.put(childName, childValue);
+ }
+
+ return map;
+ }
case ARRAY: {
final ArrayNode arrayNode = (ArrayNode) fieldNode;
final int numElements = arrayNode.size();
final Object[] arrayElements = new Object[numElements];
int count = 0;
for (final JsonNode node : arrayNode) {
- final DataType elementType;
- if (desiredType instanceof ArrayDataType) {
- elementType = ((ArrayDataType) desiredType).getElementType();
- } else {
- elementType = determineFieldType(node);
- }
-
+ final DataType elementType = ((ArrayDataType) desiredType).getElementType();
final Object converted = convertField(node, fieldName, elementType);
arrayElements[count++] = converted;
}
@@ -124,14 +151,24 @@ public class JsonTreeRowRecordReader extends AbstractJsonRowRecordReader {
}
case RECORD: {
if (fieldNode.isObject()) {
- final RecordSchema childSchema;
+ RecordSchema childSchema;
if (desiredType instanceof RecordDataType) {
childSchema = ((RecordDataType) desiredType).getChildSchema();
} else {
return null;
}
- return convertJsonNodeToRecord(fieldNode, childSchema);
+ if (childSchema == null) {
+ final List<RecordField> fields = new ArrayList<>();
+ final Iterator<String> fieldNameItr = fieldNode.getFieldNames();
+ while (fieldNameItr.hasNext()) {
+ fields.add(new RecordField(fieldNameItr.next(), RecordFieldType.STRING.getDataType()));
+ }
+
+ childSchema = new SimpleRecordSchema(fields);
+ }
+
+ return convertJsonNodeToRecord(fieldNode, childSchema, fieldName + ".");
} else {
return null;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
index 05895d8..943e1d5 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
@@ -22,17 +22,22 @@ import java.io.OutputStream;
import java.math.BigInteger;
import java.sql.SQLException;
import java.util.Collections;
+import java.util.Map;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.schema.access.SchemaAccessWriter;
import org.apache.nifi.serialization.RecordSetWriter;
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.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.RecordSet;
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.stream.io.NonCloseableOutputStream;
import org.codehaus.jackson.JsonFactory;
@@ -42,25 +47,32 @@ import org.codehaus.jackson.JsonGenerator;
public class WriteJsonResult implements RecordSetWriter {
private final ComponentLog logger;
private final boolean prettyPrint;
+ private final SchemaAccessWriter schemaAccess;
+ private final RecordSchema recordSchema;
private final JsonFactory factory = new JsonFactory();
private final String dateFormat;
private final String timeFormat;
private final String timestampFormat;
- public WriteJsonResult(final ComponentLog logger, final boolean prettyPrint, final String dateFormat, final String timeFormat, final String timestampFormat) {
+ public WriteJsonResult(final ComponentLog logger, final RecordSchema recordSchema, final SchemaAccessWriter schemaAccess, final boolean prettyPrint,
+ final String dateFormat, final String timeFormat, final String timestampFormat) {
+
+ this.logger = logger;
+ this.recordSchema = recordSchema;
this.prettyPrint = prettyPrint;
+ this.schemaAccess = schemaAccess;
this.dateFormat = dateFormat;
this.timeFormat = timeFormat;
this.timestampFormat = timestampFormat;
-
- this.logger = logger;
}
@Override
public WriteResult write(final RecordSet rs, final OutputStream rawOut) throws IOException {
int count = 0;
+ schemaAccess.writeHeader(recordSchema, rawOut);
+
try (final JsonGenerator generator = factory.createJsonGenerator(new NonCloseableOutputStream(rawOut))) {
if (prettyPrint) {
generator.useDefaultPrettyPrinter();
@@ -71,7 +83,7 @@ public class WriteJsonResult implements RecordSetWriter {
Record record;
while ((record = rs.next()) != null) {
count++;
- writeRecord(record, generator, g -> g.writeStartObject(), g -> g.writeEndObject());
+ writeRecord(record, recordSchema, generator, g -> g.writeStartObject(), g -> g.writeEndObject());
}
generator.writeEndArray();
@@ -79,7 +91,7 @@ public class WriteJsonResult implements RecordSetWriter {
throw new IOException("Failed to serialize Result Set to stream", e);
}
- return WriteResult.of(count, Collections.emptyMap());
+ return WriteResult.of(count, schemaAccess.getAttributes(recordSchema));
}
@Override
@@ -89,7 +101,7 @@ public class WriteJsonResult implements RecordSetWriter {
generator.useDefaultPrettyPrinter();
}
- writeRecord(record, generator, g -> g.writeStartObject(), g -> g.writeEndObject());
+ writeRecord(record, recordSchema, generator, g -> g.writeStartObject(), g -> g.writeEndObject());
} catch (final SQLException e) {
throw new IOException("Failed to write records to stream", e);
}
@@ -97,24 +109,24 @@ public class WriteJsonResult implements RecordSetWriter {
return WriteResult.of(1, Collections.emptyMap());
}
- private void writeRecord(final Record record, final JsonGenerator generator, final GeneratorTask startTask, final GeneratorTask endTask)
+ private void writeRecord(final Record record, final RecordSchema writeSchema, final JsonGenerator generator, final GeneratorTask startTask, final GeneratorTask endTask)
throws JsonGenerationException, IOException, SQLException {
try {
- final RecordSchema schema = record.getSchema();
startTask.apply(generator);
- for (int i = 0; i < schema.getFieldCount(); i++) {
- final String fieldName = schema.getField(i).getFieldName();
- final Object value = record.getValue(fieldName);
+ for (int i = 0; i < writeSchema.getFieldCount(); i++) {
+ final RecordField field = writeSchema.getField(i);
+ final String fieldName = field.getFieldName();
+ final Object value = record.getValue(field);
if (value == null) {
generator.writeNullField(fieldName);
continue;
}
generator.writeFieldName(fieldName);
- final DataType dataType = schema.getDataType(fieldName).get();
+ final DataType dataType = writeSchema.getDataType(fieldName).get();
- writeValue(generator, value, dataType, i < schema.getFieldCount() - 1);
+ writeValue(generator, value, fieldName, dataType, i < writeSchema.getFieldCount() - 1);
}
endTask.apply(generator);
@@ -125,7 +137,8 @@ public class WriteJsonResult implements RecordSetWriter {
}
- private void writeValue(final JsonGenerator generator, final Object value, final DataType dataType, final boolean moreCols)
+ @SuppressWarnings("unchecked")
+ private void writeValue(final JsonGenerator generator, final Object value, final String fieldName, final DataType dataType, final boolean moreCols)
throws JsonGenerationException, IOException, SQLException {
if (value == null) {
generator.writeNull();
@@ -133,7 +146,7 @@ public class WriteJsonResult implements RecordSetWriter {
}
final DataType chosenDataType = dataType.getFieldType() == RecordFieldType.CHOICE ? DataTypeUtils.chooseDataType(value, (ChoiceDataType) dataType) : dataType;
- final Object coercedValue = DataTypeUtils.convertType(value, chosenDataType);
+ final Object coercedValue = DataTypeUtils.convertType(value, chosenDataType, fieldName);
if (coercedValue == null) {
generator.writeNull();
return;
@@ -146,18 +159,18 @@ public class WriteJsonResult implements RecordSetWriter {
generator.writeString(DataTypeUtils.toString(coercedValue, dateFormat, timeFormat, timestampFormat));
break;
case DOUBLE:
- generator.writeNumber(DataTypeUtils.toDouble(coercedValue));
+ generator.writeNumber(DataTypeUtils.toDouble(coercedValue, fieldName));
break;
case FLOAT:
- generator.writeNumber(DataTypeUtils.toFloat(coercedValue));
+ generator.writeNumber(DataTypeUtils.toFloat(coercedValue, fieldName));
break;
case LONG:
- generator.writeNumber(DataTypeUtils.toLong(coercedValue));
+ generator.writeNumber(DataTypeUtils.toLong(coercedValue, fieldName));
break;
case INT:
case BYTE:
case SHORT:
- generator.writeNumber(DataTypeUtils.toInteger(coercedValue));
+ generator.writeNumber(DataTypeUtils.toInteger(coercedValue, fieldName));
break;
case CHAR:
case STRING:
@@ -182,7 +195,24 @@ public class WriteJsonResult implements RecordSetWriter {
break;
case RECORD: {
final Record record = (Record) coercedValue;
- writeRecord(record, generator, gen -> gen.writeStartObject(), gen -> gen.writeEndObject());
+ final RecordDataType recordDataType = (RecordDataType) chosenDataType;
+ final RecordSchema childSchema = recordDataType.getChildSchema();
+ writeRecord(record, childSchema, generator, gen -> gen.writeStartObject(), gen -> gen.writeEndObject());
+ break;
+ }
+ case MAP: {
+ final MapDataType mapDataType = (MapDataType) chosenDataType;
+ final DataType valueDataType = mapDataType.getValueType();
+ final Map<String, ?> map = (Map<String, ?>) coercedValue;
+ generator.writeStartObject();
+ int i = 0;
+ for (final Map.Entry<String, ?> entry : map.entrySet()) {
+ final String mapKey = entry.getKey();
+ final Object mapValue = entry.getValue();
+ generator.writeFieldName(mapKey);
+ writeValue(generator, mapValue, fieldName + "." + mapKey, valueDataType, ++i < map.size());
+ }
+ generator.writeEndObject();
break;
}
case ARRAY:
@@ -191,7 +221,7 @@ public class WriteJsonResult implements RecordSetWriter {
final Object[] values = (Object[]) coercedValue;
final ArrayDataType arrayDataType = (ArrayDataType) dataType;
final DataType elementType = arrayDataType.getElementType();
- writeArray(values, generator, elementType);
+ writeArray(values, fieldName, generator, elementType);
} else {
generator.writeString(coercedValue.toString());
}
@@ -199,12 +229,13 @@ public class WriteJsonResult implements RecordSetWriter {
}
}
- private void writeArray(final Object[] values, final JsonGenerator generator, final DataType elementType) throws JsonGenerationException, IOException, SQLException {
+ private void writeArray(final Object[] values, final String fieldName, final JsonGenerator generator, final DataType elementType)
+ throws JsonGenerationException, IOException, SQLException {
generator.writeStartArray();
for (int i = 0; i < values.length; i++) {
final boolean moreEntries = i < values.length - 1;
final Object element = values[i];
- writeValue(generator, element, elementType, moreEntries);
+ writeValue(generator, element, fieldName, elementType, moreEntries);
}
generator.writeEndArray();
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/schema/access/AvroSchemaTextStrategy.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/schema/access/AvroSchemaTextStrategy.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/schema/access/AvroSchemaTextStrategy.java
new file mode 100644
index 0000000..27f84e4
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/schema/access/AvroSchemaTextStrategy.java
@@ -0,0 +1,64 @@
+/*
+ * 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.schema.access;
+
+import java.io.InputStream;
+import java.util.EnumSet;
+import java.util.Set;
+
+import org.apache.avro.Schema;
+import org.apache.nifi.avro.AvroTypeUtil;
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AvroSchemaTextStrategy implements SchemaAccessStrategy {
+ private static final Set<SchemaField> schemaFields = EnumSet.of(SchemaField.SCHEMA_TEXT, SchemaField.SCHEMA_TEXT_FORMAT);
+
+ private static final Logger logger = LoggerFactory.getLogger(AvroSchemaTextStrategy.class);
+ private final PropertyValue schemaTextPropertyValue;
+
+ public AvroSchemaTextStrategy(final PropertyValue schemaTextPropertyValue) {
+ this.schemaTextPropertyValue = schemaTextPropertyValue;
+ }
+
+ @Override
+ public RecordSchema getSchema(final FlowFile flowFile, final InputStream contentStream, final ConfigurationContext context) throws SchemaNotFoundException {
+ final String schemaText = schemaTextPropertyValue.evaluateAttributeExpressions(flowFile).getValue();
+ if (schemaText == null || schemaText.trim().isEmpty()) {
+ throw new SchemaNotFoundException("FlowFile did not contain appropriate attributes to determine Schema Text");
+ }
+
+ logger.debug("For {} found schema text {}", flowFile, schemaText);
+
+ try {
+ final Schema avroSchema = new Schema.Parser().parse(schemaText);
+ return AvroTypeUtil.createSchema(avroSchema);
+ } catch (final Exception e) {
+ throw new SchemaNotFoundException("Failed to create schema from the Schema Text after evaluating FlowFile Attributes", e);
+ }
+ }
+
+ @Override
+ public Set<SchemaField> getSuppliedSchemaFields() {
+ return schemaFields;
+ }
+}