@Test public void testMerge() throws InvalidProtocolBufferException { List<byte[]> list = Arrays.asList(b1, b2); Iterator it = list.iterator(); List<byte[]> list2 = Arrays.asList(b3, b4); Iterator it2 = list2.iterator(); byte[] a1 = example.aggregate(new HashMap<>(), it); byte[] a2 = example.aggregate(new HashMap<>(), it2); Iterator mergedIt = Arrays.asList(a1, a2).iterator(); byte[] mergedByteArray = example.merge(new HashMap<>(), mergedIt); List<SdiiMessage.Message> mergedMessageList = SdiiMessageList.MessageList.parseFrom(mergedByteArray).getMessageList(); assertEquals(m1, mergedMessageList.get(0)); assertEquals(m2, mergedMessageList.get(1)); assertEquals(m3, mergedMessageList.get(2)); assertEquals(m4, mergedMessageList.get(3)); }
@Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> mergedList = new ArrayList<>(); while (files.hasNext()) { byte[] data = files.next(); try { SdiiMessageList.MessageList messageListProtoBuf = SdiiMessageList.MessageList.parseFrom(data); List<SdiiMessage.Message> list = messageListProtoBuf.getMessageList(); mergedList.addAll(list); } catch (InvalidProtocolBufferException e) { LOG.error("Merge errors....", e); return null; } } return SdiiMessageList.MessageList.newBuilder().addAllMessage(mergedList).build().toByteArray(); }
ProtobufSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> mergedList = new ArrayList<>(); while (files.hasNext()) { byte[] data = files.next(); try { SdiiMessageList.MessageList messageListProtoBuf = SdiiMessageList.MessageList.parseFrom(data); List<SdiiMessage.Message> list = messageListProtoBuf.getMessageList(); mergedList.addAll(list); } catch (InvalidProtocolBufferException e) { LOG.error("Merge errors....", e); return null; } } return SdiiMessageList.MessageList.newBuilder().addAllMessage(mergedList).build().toByteArray(); } }
ProtobufSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> mergedList = new ArrayList<>(); while (files.hasNext()) { byte[] data = files.next(); try { SdiiMessageList.MessageList messageListProtoBuf = SdiiMessageList.MessageList.parseFrom(data); List<SdiiMessage.Message> list = messageListProtoBuf.getMessageList(); mergedList.addAll(list); } catch (InvalidProtocolBufferException e) { LOG.error("Merge errors....", e); return null; } } return SdiiMessageList.MessageList.newBuilder().addAllMessage(mergedList).build().toByteArray(); }  }
ProtobufSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> mergedList = new ArrayList<>(); while (files.hasNext()) { byte[] data = files.next(); try { SdiiMessageList.MessageList messageListProtoBuf = SdiiMessageList.MessageList.parseFrom(data); List<SdiiMessage.Message> list = messageListProtoBuf.getMessageList(); mergedList.addAll(list); } catch (InvalidProtocolBufferException e) { LOG.error("Merge errors....", e); return null; } } return SdiiMessageList.MessageList.newBuilder().addAllMessage(mergedList).build().toByteArray(); }  @Override Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files); }
ProtobufSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> mergedList = new ArrayList<>(); while (files.hasNext()) { byte[] data = files.next(); try { SdiiMessageList.MessageList messageListProtoBuf = SdiiMessageList.MessageList.parseFrom(data); List<SdiiMessage.Message> list = messageListProtoBuf.getMessageList(); mergedList.addAll(list); } catch (InvalidProtocolBufferException e) { LOG.error("Merge errors....", e); return null; } } return SdiiMessageList.MessageList.newBuilder().addAllMessage(mergedList).build().toByteArray(); }  @Override Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files);  }