Repository: nifi Updated Branches: refs/heads/master 921403ff5 -> 022b64be0
NIFI-5316 Fixed array handling for Avro that comes from Parquet's Avro reader Signed-off-by: zenfenan <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/022b64be Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/022b64be Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/022b64be Branch: refs/heads/master Commit: 022b64be08f1c9a178f0fb917402c10bc7f4ec9b Parents: 921403f Author: Bryan Bende <[email protected]> Authored: Mon Jun 25 10:36:55 2018 -0400 Committer: zenfenan <[email protected]> Committed: Thu Jun 28 22:13:23 2018 +0530 ---------------------------------------------------------------------- .../java/org/apache/nifi/avro/AvroTypeUtil.java | 22 +++- .../nifi-parquet-processors/pom.xml | 2 + .../processors/parquet/FetchParquetTest.java | 110 +++++++++++++++++++ .../test/resources/avro/user-with-array.avsc | 9 ++ .../avro/user-with-nullable-array.avsc | 9 ++ 5 files changed, 146 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/022b64be/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java index c214819..23f74b8 100755 --- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java +++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java @@ -894,13 +894,23 @@ public class AvroTypeUtil { case STRING: return value.toString(); case ARRAY: - final GenericData.Array<?> array = (GenericData.Array<?>) value; - final Object[] valueArray = new Object[array.size()]; - for (int i = 0; i < array.size(); i++) { - final Schema elementSchema = avroSchema.getElementType(); - valueArray[i] = normalizeValue(array.get(i), elementSchema, fieldName + "[" + i + "]"); + if (value instanceof List) { + final List<?> list = (List<?>) value; + final Object[] valueArray = new Object[list.size()]; + for (int i = 0; i < list.size(); i++) { + final Schema elementSchema = avroSchema.getElementType(); + valueArray[i] = normalizeValue(list.get(i), elementSchema, fieldName + "[" + i + "]"); + } + return valueArray; + } else { + final GenericData.Array<?> array = (GenericData.Array<?>) value; + final Object[] valueArray = new Object[array.size()]; + for (int i = 0; i < array.size(); i++) { + final Schema elementSchema = avroSchema.getElementType(); + valueArray[i] = normalizeValue(array.get(i), elementSchema, fieldName + "[" + i + "]"); + } + return valueArray; } - return valueArray; case MAP: final Map<?, ?> avroMap = (Map<?, ?>) value; final Map<String, Object> map = new HashMap<>(avroMap.size()); http://git-wip-us.apache.org/repos/asf/nifi/blob/022b64be/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/pom.xml b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/pom.xml index 5c9c19e..8a76c68 100644 --- a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/pom.xml +++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/pom.xml @@ -101,6 +101,8 @@ <configuration> <excludes combine.children="append"> <exclude>src/test/resources/avro/user.avsc</exclude> + <exclude>src/test/resources/avro/user-with-array.avsc</exclude> + <exclude>src/test/resources/avro/user-with-nullable-array.avsc</exclude> </excludes> </configuration> </plugin> http://git-wip-us.apache.org/repos/asf/nifi/blob/022b64be/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java index 76d44aa..83b11f2 100644 --- a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java +++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java @@ -62,6 +62,8 @@ public class FetchParquetTest { static final String RECORD_HEADER = "name,favorite_number,favorite_color"; private Schema schema; + private Schema schemaWithArray; + private Schema schemaWithNullableArray; private Configuration testConf; private FetchParquet proc; private TestRunner testRunner; @@ -71,6 +73,12 @@ public class FetchParquetTest { final String avroSchema = IOUtils.toString(new FileInputStream("src/test/resources/avro/user.avsc"), StandardCharsets.UTF_8); schema = new Schema.Parser().parse(avroSchema); + final String avroSchemaWithArray = IOUtils.toString(new FileInputStream("src/test/resources/avro/user-with-array.avsc"), StandardCharsets.UTF_8); + schemaWithArray = new Schema.Parser().parse(avroSchemaWithArray); + + final String avroSchemaWithNullableArray = IOUtils.toString(new FileInputStream("src/test/resources/avro/user-with-nullable-array.avsc"), StandardCharsets.UTF_8); + schemaWithNullableArray = new Schema.Parser().parse(avroSchemaWithNullableArray); + testConf = new Configuration(); testConf.addResource(new Path(TEST_CONF_PATH)); @@ -243,6 +251,42 @@ public class FetchParquetTest { flowFile.assertContentEquals("TRIGGER"); } + @Test + public void testFetchWithArray() throws InitializationException, IOException { + configure(proc); + + final File parquetDir = new File(DIRECTORY); + final File parquetFile = new File(parquetDir,"testFetchParquetWithArrayToCSV.parquet"); + final int numUsers = 10; + writeParquetUsersWithArray(parquetFile, numUsers); + + final Map<String,String> attributes = new HashMap<>(); + attributes.put(CoreAttributes.PATH.key(), parquetDir.getAbsolutePath()); + attributes.put(CoreAttributes.FILENAME.key(), parquetFile.getName()); + + testRunner.enqueue("TRIGGER", attributes); + testRunner.run(); + testRunner.assertAllFlowFilesTransferred(FetchParquet.REL_SUCCESS, 1); + } + + @Test + public void testFetchWithNullableArray() throws InitializationException, IOException { + configure(proc); + + final File parquetDir = new File(DIRECTORY); + final File parquetFile = new File(parquetDir,"testFetchParquetWithNullableArrayToCSV.parquet"); + final int numUsers = 10; + writeParquetUsersWithNullableArray(parquetFile, numUsers); + + final Map<String,String> attributes = new HashMap<>(); + attributes.put(CoreAttributes.PATH.key(), parquetDir.getAbsolutePath()); + attributes.put(CoreAttributes.FILENAME.key(), parquetFile.getName()); + + testRunner.enqueue("TRIGGER", attributes); + testRunner.run(); + testRunner.assertAllFlowFilesTransferred(FetchParquet.REL_SUCCESS, 1); + } + protected void verifyCSVRecords(int numUsers, String csvContent) { final String[] splits = csvContent.split("[\\n]"); Assert.assertEquals(numUsers, splits.length); @@ -278,4 +322,70 @@ public class FetchParquetTest { } + private void writeParquetUsersWithArray(final File parquetFile, int numUsers) throws IOException { + if (parquetFile.exists()) { + Assert.assertTrue(parquetFile.delete()); + } + + final Path parquetPath = new Path(parquetFile.getPath()); + + final AvroParquetWriter.Builder<GenericRecord> writerBuilder = AvroParquetWriter + .<GenericRecord>builder(parquetPath) + .withSchema(schemaWithArray) + .withConf(testConf); + + final Schema favoriteColorsSchema = schemaWithArray.getField("favorite_colors").schema(); + + try (final ParquetWriter<GenericRecord> writer = writerBuilder.build()) { + for (int i=0; i < numUsers; i++) { + final GenericRecord user = new GenericData.Record(schema); + user.put("name", "Bob" + i); + user.put("favorite_number", i); + + + final GenericData.Array<String> colors = new GenericData.Array<>(1, favoriteColorsSchema); + colors.add("blue" + i); + + user.put("favorite_color", colors); + + writer.write(user); + } + } + + } + + private void writeParquetUsersWithNullableArray(final File parquetFile, int numUsers) throws IOException { + if (parquetFile.exists()) { + Assert.assertTrue(parquetFile.delete()); + } + + final Path parquetPath = new Path(parquetFile.getPath()); + + final AvroParquetWriter.Builder<GenericRecord> writerBuilder = AvroParquetWriter + .<GenericRecord>builder(parquetPath) + .withSchema(schemaWithNullableArray) + .withConf(testConf); + + // use the schemaWithArray here just to get the schema for the array part of the favorite_colors fields, the overall + // schemaWithNullableArray has a union of the array schema and null + final Schema favoriteColorsSchema = schemaWithArray.getField("favorite_colors").schema(); + + try (final ParquetWriter<GenericRecord> writer = writerBuilder.build()) { + for (int i=0; i < numUsers; i++) { + final GenericRecord user = new GenericData.Record(schema); + user.put("name", "Bob" + i); + user.put("favorite_number", i); + + + final GenericData.Array<String> colors = new GenericData.Array<>(1, favoriteColorsSchema); + colors.add("blue" + i); + + user.put("favorite_color", colors); + + writer.write(user); + } + } + + } + } http://git-wip-us.apache.org/repos/asf/nifi/blob/022b64be/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-array.avsc ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-array.avsc b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-array.avsc new file mode 100644 index 0000000..67a0cca --- /dev/null +++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-array.avsc @@ -0,0 +1,9 @@ +{"namespace": "example.avro", + "type": "record", + "name": "User", + "fields": [ + {"name": "name", "type": "string"}, + {"name": "favorite_number", "type": ["int", "null"]}, + {"name": "favorite_colors", "type": { "type": "array", "items": ["string","null"] }, "default": null } + ] +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/022b64be/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-nullable-array.avsc ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-nullable-array.avsc b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-nullable-array.avsc new file mode 100644 index 0000000..8986eba --- /dev/null +++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/resources/avro/user-with-nullable-array.avsc @@ -0,0 +1,9 @@ +{"namespace": "example.avro", + "type": "record", + "name": "User", + "fields": [ + {"name": "name", "type": "string"}, + {"name": "favorite_number", "type": ["int", "null"]}, + {"name": "favorite_colors", "type": [ "null", { "type": "array", "items": ["string","null"] } ], "default": null } + ] +} \ No newline at end of file
