[
https://issues.apache.org/jira/browse/NIFI-751?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14618995#comment-14618995
]
ASF GitHub Bot commented on NIFI-751:
-------------------------------------
Github user rdblue commented on a diff in the pull request:
https://github.com/apache/incubator-nifi/pull/70#discussion_r34174772
--- Diff:
nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertAvroSchema.java
---
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.nifi.processors.kite;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.avro.Schema;
+import org.apache.avro.file.CodecFactory;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData.Record;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.io.StreamCallback;
+import
org.apache.nifi.processors.kite.AvroRecordConverter.AvroConversionException;
+import org.apache.nifi.util.LongHolder;
+import org.codehaus.jackson.JsonNode;
+import org.kitesdk.data.DatasetException;
+import org.kitesdk.data.DatasetIOException;
+import org.kitesdk.data.SchemaNotFoundException;
+import org.kitesdk.data.spi.DefaultConfiguration;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+
+@Tags({ "kite", "avro" })
+@CapabilityDescription("Convert records from one Avro schema to another,
including support for flattening and simple type conversions")
+public class ConvertAvroSchema extends AbstractKiteProcessor {
+
+ private static final Relationship SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Avro content that was converted
successfully")
+ .build();
+
+ private static final Relationship FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description(
+ "Avro content that could not be
converterted to output schema")
+ .build();
+
+ private static final Relationship INCOMPATIBLE = new
Relationship.Builder()
+ .name("incompatible")
+ .description(
+ "Avro content that could not be read
with input schema")
+ .build();
+
+ @VisibleForTesting
+ static final PropertyDescriptor INPUT_SCHEMA = new
PropertyDescriptor.Builder()
+ .name("Input Schema").description("Avro Schema of Input
Flowfiles")
+
.addValidator(SCHEMA_VALIDATOR).expressionLanguageSupported(true)
+ .required(true).build();
+
+ @VisibleForTesting
+ static final PropertyDescriptor OUTPUT_SCHEMA = new
PropertyDescriptor.Builder()
+ .name("Output Schema")
+ .description("Avro Schema of Output Flowfiles")
+
.addValidator(SCHEMA_VALIDATOR).expressionLanguageSupported(true)
+ .required(true).build();
+
+ @VisibleForTesting
+ static final PropertyDescriptor FIELD_MAPPING = new
PropertyDescriptor.Builder()
+ .name("Field Mapping")
+ .description(
+ "Field Mapping Between Schemas. Format
is a JSON array of maps with properties "
+ + "source (name of
column in input record, including . notation to step into "
+ + "records) and target
(name of column in output record). For fields not "
+ + "found in this
mapping, the processor tries to match names from the "
+ + "input to the output
record.")
+
.addValidator(JSON_VALIDATOR).defaultValue("[]").build();
+
+ private static final List<PropertyDescriptor> PROPERTIES = ImmutableList
+ .<PropertyDescriptor> builder()
+
.addAll(AbstractKiteProcessor.getProperties()).add(INPUT_SCHEMA)
+ .add(OUTPUT_SCHEMA).add(FIELD_MAPPING).build();
+
+ private static final Set<Relationship> RELATIONSHIPS = ImmutableSet
+ .<Relationship> builder().add(SUCCESS).add(FAILURE)
+ .add(INCOMPATIBLE).build();
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, final ProcessSession
session)
+ throws ProcessException {
+ FlowFile incomingAvro = session.get();
+ if (incomingAvro == null) {
+ return;
+ }
+
+ String inputSchemaProperty = context.getProperty(INPUT_SCHEMA)
+
.evaluateAttributeExpressions(incomingAvro).getValue();
+ final Schema inputSchema;
+ try {
+ inputSchema = getSchema(inputSchemaProperty,
+ DefaultConfiguration.get());
+ } catch (SchemaNotFoundException e) {
+ getLogger().error("Cannot find schema: " +
inputSchemaProperty);
+ session.transfer(incomingAvro, FAILURE);
+ return;
+ }
+ String outputSchemaProperty = context.getProperty(OUTPUT_SCHEMA)
+
.evaluateAttributeExpressions(incomingAvro).getValue();
+ final Schema outputSchema;
+ try {
+ outputSchema = getSchema(outputSchemaProperty,
+ DefaultConfiguration.get());
+ } catch (SchemaNotFoundException e) {
+ getLogger().error("Cannot find schema: " +
outputSchemaProperty);
+ session.transfer(incomingAvro, FAILURE);
+ return;
+ }
+ String fieldMappingProperty = context.getProperty(FIELD_MAPPING)
+
.evaluateAttributeExpressions(incomingAvro).getValue();
+ final JsonNode fieldMapping;
+ try {
+ fieldMapping = getJSON(fieldMappingProperty,
+ DefaultConfiguration.get());
+ } catch (IllegalArgumentException e) {
+ getLogger().error(
+ "Cannot find field mapping: " +
fieldMappingProperty);
+ session.transfer(incomingAvro, FAILURE);
+ return;
+ }
+ final AvroRecordConverter converter = new AvroRecordConverter(
+ inputSchema, outputSchema, fieldMapping);
+
+ final DataFileWriter<Record> writer = new DataFileWriter<>(
+ AvroUtil.newDatumWriter(outputSchema,
Record.class));
+ writer.setCodec(CodecFactory.snappyCodec());
+
+ try {
+ final LongHolder written = new LongHolder(0L);
+ final FailureTracker failures = new FailureTracker();
+
+ FlowFile badRecords = session.clone(incomingAvro);
+ FlowFile outgoingAvro = session.write(incomingAvro,
+ new StreamCallback() {
+ @Override
+ public void process(InputStream
in, OutputStream out)
+ throws
IOException {
+ try
(DataFileStream<Record> stream = new DataFileStream<Record>(
+ in, new
GenericDatumReader<Record>(
+
converter.getInputSchema()))) {
+ try
(DataFileWriter<Record> w = writer.create(
+
outputSchema, out)) {
+ for
(Record record : stream) {
+
try {
+
Record converted = converter
+
.convert(record);
+
w.append(converted);
+
written.incrementAndGet();
+
} catch (AvroConversionException e) {
+
failures.add(e);
--- End diff --
This doesn't have to go in the initial version, but since we have the
original schema and the original datum, we can send just the failed records on
the "incompatible" relationship and not the entire file. I think that would be
valuable, but I can add that when I add it to CSV and JSON.
> Add Processor To Convert Avro Formats
> -------------------------------------
>
> Key: NIFI-751
> URL: https://issues.apache.org/jira/browse/NIFI-751
> Project: Apache NiFi
> Issue Type: Improvement
> Components: Extensions
> Affects Versions: 0.1.0
> Reporter: Alan Jackoway
>
> When working with data from external sources, such as complex WSDL, I
> frequently wind up with complex nested data that is difficult to work with
> even when converted to Avro format. Specifically, I often have two needs:
> * Converting types of data, usually from string to long, double, etc. when
> APIs give only string data back.
> * Flattening data by taking fields out of nested records and putting them on
> the top level of the Avro file.
> Unfortunately the Kite JSONToAvro processor only supports exact conversions
> from JSON to a matching Avro schema and will not do data transformations of
> this type. Proposed processor to come.
> Discussed this with [~rdblue], so tagging him here as I don't have permission
> to set a CC for some reason.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
