@Test public void testMerge() throws IOException { SdiiMessage.Message message1 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d); SdiiMessage.Message message2 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d); SdiiMessage.Message message3 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d); SdiiMessage.Message message4 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d); example.open(parameters, runtimeContext); byte[] archiveBytesMessage1 = example.aggregate( null, Arrays.asList(message1.toByteArray(), message2.toByteArray()).iterator()); byte[] archiveBytesMessage2 = example.aggregate( null, Arrays.asList(message3.toByteArray(), message4.toByteArray()).iterator()); File tmpFile = File.createTempFile("test", ".avro"); tmpFile.deleteOnExit(); FileUtils.writeByteArrayToFile( tmpFile, example.merge( new HashMap<>(), Arrays.asList(archiveBytesMessage1, archiveBytesMessage2).iterator())); List<SdiiMessage.Message> list = AvroHelper.fromFile(tmpFile, SdiiMessage.Message.class); assertEquals(4, list.size()); assertSDIIMessagesAreEqual(message1, list.get(0)); assertSDIIMessagesAreEqual(message2, list.get(1)); assertSDIIMessagesAreEqual(message3, list.get(2)); assertSDIIMessagesAreEqual(message4, list.get(3)); }
@Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { try { return AvroHelper.mergeAvroFiles(files, SdiiMessage.Message.class); } catch (Exception e) { LOG.error("Merge errors....", e); } return null; }
AvroSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { try { return AvroHelper.mergeAvroFiles(files, SdiiMessage.Message.class); } catch (Exception e) { LOG.error("Merge errors....", e); } return null; } }
AvroSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { try { return AvroHelper.mergeAvroFiles(files, SdiiMessage.Message.class); } catch (Exception e) { LOG.error("Merge errors....", e); } return null; }  }
AvroSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { try { return AvroHelper.mergeAvroFiles(files, SdiiMessage.Message.class); } catch (Exception e) { LOG.error("Merge errors....", e); } return null; }  @Override void open(Configuration parameters, RuntimeContext runtimeContext); @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); }
AvroSimpleKeyExample implements SimpleUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { try { return AvroHelper.mergeAvroFiles(files, SdiiMessage.Message.class); } catch (Exception e) { LOG.error("Merge errors....", e); } return null; }  @Override void open(Configuration parameters, RuntimeContext runtimeContext); @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);  }