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; + } +}
