Github user rdblue commented on a diff in the pull request:
https://github.com/apache/incubator-nifi/pull/70#discussion_r34175017
--- 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);
+
getLogger().error(
+
"Error converting data: "
+
+ e.getMessage());
+
}
+ }
+ }
+ }
+ }
+ });
+
+ long errors = failures.count();
+
+ // update only if file transfer is successful
+ session.adjustCounter("Converted records",
written.get(), false);
+ // update only if file transfer is successful
+ session.adjustCounter("Conversion errors", errors,
false);
+
+ if (written.get() > 0L) {
+ session.transfer(outgoingAvro, SUCCESS);
+
+ if (errors > 0L) {
+ getLogger().warn(
+ "Failed to convert
{}/{} records from CSV to Avro",
+ new Object[] { errors,
errors + written.get() });
+ badRecords =
session.putAttribute(badRecords, "errors",
+ failures.summary());
+ session.transfer(badRecords,
INCOMPATIBLE);
+ } else {
+ session.remove(badRecords);
+ }
+
+ } else {
+ session.remove(outgoingAvro);
+
+ if (errors > 0L) {
+ getLogger().warn(
+ "Failed to convert
{}/{} records from CSV to Avro",
+ new Object[] { errors,
errors });
+ badRecords =
session.putAttribute(badRecords, "errors",
+ failures.summary());
+ } else {
+ badRecords =
session.putAttribute(badRecords, "errors",
+ "No incoming records");
+ }
+
+ session.transfer(badRecords, FAILURE);
+ }
+
+ } catch (ProcessException | DatasetIOException e) {
+ getLogger().error("Failed reading or writing", e);
+ session.transfer(incomingAvro, FAILURE);
+ } catch (DatasetException e) {
+ getLogger().error("Failed to read FlowFile", e);
+ session.transfer(incomingAvro, FAILURE);
+ } finally {
+ try {
+ writer.close();
--- End diff --
I think this is handled by the try-with-resources block in the
`StreamCallback`. The writer should be closed correctly after that so there is
no need to do it here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
