ProtobufUtils { public static List<Object> buildRowValues(Descriptors.Descriptor dsc, Message msg, StructType schema) { List<Object> values = new ArrayList<>(); Object val; for (Descriptors.FieldDescriptor fd : dsc.getFields()) { if ( (!fd.isRepeated() && msg.hasField(fd)) || (fd.isRepeated() && msg.getRepeatedFieldCount(fd) > 0) ) { val = getFieldValue(fd, msg.getField(fd), schema.apply(fd.getName()).dataType()); } else { LOG.trace("FieldDescriptor[{}] => not found", fd.getFullName()); val = null; } values.add(val); } return values; }  static List<Object> buildRowValues(Descriptors.Descriptor dsc, Message msg, StructType schema); static StructType buildSchema(Descriptors.Descriptor descriptor); static Descriptors.Descriptor buildDescriptor(String filePath, String messageName); static Descriptors.Descriptor buildDescriptor(String filePath); static boolean isGzipped(InputStream inputStream);  }
@Test public void buildRowValues() { DynamicMessage nestedMessage0 = DynamicMessage.newBuilder(SINGLE_DESC.findNestedTypeByName("NestedExample")) .setField(SINGLE_DESC.findNestedTypeByName("NestedExample").findFieldByName("nested"), "a nested type") .build(); DynamicMessage nestedMessage1 = DynamicMessage.newBuilder(SINGLE_DESC.findNestedTypeByName("NestedExample")) .setField(SINGLE_DESC.findNestedTypeByName("NestedExample").findFieldByName("nested"), "another nested type") .build(); DynamicMessage mapInt0 = DynamicMessage.newBuilder(SINGLE_DESC.findNestedTypeByName("MapIntEntry")) .setField(SINGLE_DESC.findNestedTypeByName("MapIntEntry").findFieldByName("key"), "key value") .setField(SINGLE_DESC.findNestedTypeByName("MapIntEntry").findFieldByName("value"), 987) .build(); DynamicMessage mapInt1 = DynamicMessage.newBuilder(SINGLE_DESC.findNestedTypeByName("MapIntEntry")) .setField(SINGLE_DESC.findNestedTypeByName("MapIntEntry").findFieldByName("key"), "another key value") .setField(SINGLE_DESC.findNestedTypeByName("MapIntEntry").findFieldByName("value"), 123) .build(); Map<String, Integer> stringIntegerMap = new HashMap<>(); stringIntegerMap.put("key value", 987); stringIntegerMap.put("another key value", 123); DynamicMessage message = DynamicMessage.newBuilder(SINGLE_DESC) .setField(SINGLE_DESC.findFieldByName("string"), "string value") .setField(SINGLE_DESC.findFieldByName("double"), 987.654D) .setField(SINGLE_DESC.findFieldByName("float"), 987.654F) .setField(SINGLE_DESC.findFieldByName("int32"), 987) .setField(SINGLE_DESC.findFieldByName("int64"), 987L) .setField(SINGLE_DESC.findFieldByName("uint32"), 987) .setField(SINGLE_DESC.findFieldByName("uint64"), 987L) .setField(SINGLE_DESC.findFieldByName("sint32"), 987) .setField(SINGLE_DESC.findFieldByName("sint64"), 987L) .setField(SINGLE_DESC.findFieldByName("fixed32"), 987) .setField(SINGLE_DESC.findFieldByName("fixed64"), 987L) .setField(SINGLE_DESC.findFieldByName("sfixed32"), 987) .setField(SINGLE_DESC.findFieldByName("sfixed64"), 987L) .setField(SINGLE_DESC.findFieldByName("boolean"), true) .setField(SINGLE_DESC.findFieldByName("bytes"), ByteString.copyFrom("foo".getBytes())) .setField(SINGLE_DESC.findFieldByName("enum"), SINGLE_DESC.findEnumTypeByName("EnumExample") .findValueByName("THREE")) .setField(SINGLE_DESC.findFieldByName("nested"), nestedMessage0) .addRepeatedField(SINGLE_DESC.findFieldByName("map_int"), mapInt0) .addRepeatedField(SINGLE_DESC.findFieldByName("map_int"), mapInt1) .addRepeatedField(SINGLE_DESC.findFieldByName("repeating_message"), nestedMessage0) .addRepeatedField(SINGLE_DESC.findFieldByName("repeating_message"), nestedMessage1) .addRepeatedField(SINGLE_DESC.findFieldByName("repeating_int32"), 1234) .addRepeatedField(SINGLE_DESC.findFieldByName("repeating_int32"), 5678) .addRepeatedField(SINGLE_DESC.findFieldByName("repeating_enum"), SINGLE_DESC.findEnumTypeByName("EnumExample") .findValueByName("TWO")) .addRepeatedField(SINGLE_DESC.findFieldByName("repeating_enum"), SINGLE_DESC.findEnumTypeByName("EnumExample") .findValueByName("TWO")) .build(); List<Object> results = ProtobufUtils.buildRowValues(SINGLE_DESC, message, SINGLE_SCHEMA); assertThat(results, contains( (Object) "string value", 987.654D, 987.654F, 987, 987L, 987, 987L, 987, 987L, 987, 987L, 987, 987L, true, "foo".getBytes(), "THREE", RowFactory.create("a nested type"), JavaConversions.mapAsScalaMap(stringIntegerMap), JavaConversions.asScalaBuffer(Arrays.asList(RowFactory.create("a nested type"), RowFactory.create("another nested type"))), JavaConversions.asScalaBuffer(Arrays.asList(1234, 5678)), JavaConversions.asScalaBuffer(Arrays.asList("TWO", "TWO")), null, null, null ) ); }
@Test public void buildRowValuesSparse() { DynamicMessage message = DynamicMessage.newBuilder(SINGLE_DESC) .setField(SINGLE_DESC.findFieldByName("string"), "string value") .build(); List<Object> results = ProtobufUtils.buildRowValues(SINGLE_DESC, message, SINGLE_SCHEMA); assertThat(results, containsInRelativeOrder((Object) "string value", null)); assertThat(results.size(), is(24)); }
@Test public void buildRowValuesIndexed() { DynamicMessage message = DynamicMessage.newBuilder(SINGLE_DESC) .setField(SINGLE_DESC.findFieldByName("boolean"), true) .build(); List<Object> results = ProtobufUtils.buildRowValues(SINGLE_DESC, message, SINGLE_SCHEMA); assertThat(results, containsInRelativeOrder((Object) true, null)); assertThat(results.size(), is(24)); assertThat(results.indexOf((Object) true), is(13)); }