@Test public void testMerge() throws IOException { long receivedTime = System.currentTimeMillis(); double longitude = 10d; double latitude = 10d; SdiiMessage.Message sdiiMessageOne = prepareSDIIMessage(receivedTime, longitude, latitude); SdiiMessage.Message sdiiMessageTwo = prepareSDIIMessage(receivedTime, longitude + 0.01d, latitude + 0.01d); SdiiMessage.Message sdiiMessageThree = prepareSDIIMessage(receivedTime, longitude + 0.01d, latitude - 0.01d); SdiiMessage.Message sdiiMessageFour = prepareSDIIMessage(receivedTime, longitude - 0.01d, latitude - 0.01d); byte[] messageOne = sdiiMessageOne.toByteArray(); byte[] messageTwo = sdiiMessageTwo.toByteArray(); byte[] messageThree = sdiiMessageThree.toByteArray(); byte[] messageFour = sdiiMessageFour.toByteArray(); List<byte[]> messagesGroupOne = Arrays.asList(messageOne, messageTwo); Iterator<byte[]> messagesGroupOneIterator = messagesGroupOne.iterator(); List<byte[]> messagesGroupTwo = Arrays.asList(messageThree, messageFour); Iterator<byte[]> messagesGroupTwoIterator = messagesGroupTwo.iterator(); byte[] aggregatedMessagesGroupOne = aggregate(messagesGroupOneIterator); byte[] aggregatedMessagesGroupTwo = aggregate(messagesGroupTwoIterator); Iterator<byte[]> aggregatedMessagesIterator = Arrays.asList(aggregatedMessagesGroupOne, aggregatedMessagesGroupTwo).iterator(); ParquetCompactionExample parquetCompactionExample = new ParquetCompactionExample(); byte[] mergedMessages = parquetCompactionExample.merge(new HashMap<>(), aggregatedMessagesIterator); Path temporaryDirectory = Files.createTempDirectory("parquetTmp"); temporaryDirectory.toFile().deleteOnExit(); Path parquetFilePath = temporaryDirectory.resolve(UUID.randomUUID().toString()); Files.write(parquetFilePath, mergedMessages); ParquetReader<SdiiMessage.Message.Builder> reader = ProtoParquetReader.<SdiiMessage.Message.Builder>builder( new org.apache.hadoop.fs.Path(parquetFilePath.toString())) .build(); assertEquals(sdiiMessageOne, reader.read().build()); assertEquals(sdiiMessageTwo, reader.read().build()); assertEquals(sdiiMessageThree, reader.read().build()); assertEquals(sdiiMessageFour, reader.read().build()); assertNull(reader.read()); reader.close(); parquetFilePath.toFile().delete(); }
@Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { Path parquetTmpFilePath = null; Path parquetTargetFilePath = null; try { parquetTargetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetReader<SdiiMessage.Message.Builder> parquetReader; ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetTargetFilePath); while (files.hasNext()) { parquetTmpFilePath = tmpDir.resolve(UUID.randomUUID().toString()); Files.write(parquetTmpFilePath, files.next()); parquetReader = ProtoParquetReader.<SdiiMessage.Message.Builder>builder( new org.apache.hadoop.fs.Path(parquetTmpFilePath.toString())) .build(); for (SdiiMessage.Message.Builder sdiiMessageBuilder; (sdiiMessageBuilder = parquetReader.read()) != null; ) { parquetWriter.write(sdiiMessageBuilder.build()); } parquetReader.close(); deleteTempParquetFile(parquetTmpFilePath); } parquetWriter.close(); return Files.readAllBytes(parquetTargetFilePath); } catch (IOException e) { LOG.error("Error merging files (key={})", keys.values(), e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; }
ParquetCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { Path parquetTmpFilePath = null; Path parquetTargetFilePath = null; try { parquetTargetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetReader<SdiiMessage.Message.Builder> parquetReader; ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetTargetFilePath); while (files.hasNext()) { parquetTmpFilePath = tmpDir.resolve(UUID.randomUUID().toString()); Files.write(parquetTmpFilePath, files.next()); parquetReader = ProtoParquetReader.<SdiiMessage.Message.Builder>builder( new org.apache.hadoop.fs.Path(parquetTmpFilePath.toString())) .build(); for (SdiiMessage.Message.Builder sdiiMessageBuilder; (sdiiMessageBuilder = parquetReader.read()) != null; ) { parquetWriter.write(sdiiMessageBuilder.build()); } parquetReader.close(); deleteTempParquetFile(parquetTmpFilePath); } parquetWriter.close(); return Files.readAllBytes(parquetTargetFilePath); } catch (IOException e) { LOG.error("Error merging files (key={})", keys.values(), e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; } }
ParquetCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { Path parquetTmpFilePath = null; Path parquetTargetFilePath = null; try { parquetTargetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetReader<SdiiMessage.Message.Builder> parquetReader; ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetTargetFilePath); while (files.hasNext()) { parquetTmpFilePath = tmpDir.resolve(UUID.randomUUID().toString()); Files.write(parquetTmpFilePath, files.next()); parquetReader = ProtoParquetReader.<SdiiMessage.Message.Builder>builder( new org.apache.hadoop.fs.Path(parquetTmpFilePath.toString())) .build(); for (SdiiMessage.Message.Builder sdiiMessageBuilder; (sdiiMessageBuilder = parquetReader.read()) != null; ) { parquetWriter.write(sdiiMessageBuilder.build()); } parquetReader.close(); deleteTempParquetFile(parquetTmpFilePath); } parquetWriter.close(); return Files.readAllBytes(parquetTargetFilePath); } catch (IOException e) { LOG.error("Error merging files (key={})", keys.values(), e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; }  }
ParquetCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { Path parquetTmpFilePath = null; Path parquetTargetFilePath = null; try { parquetTargetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetReader<SdiiMessage.Message.Builder> parquetReader; ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetTargetFilePath); while (files.hasNext()) { parquetTmpFilePath = tmpDir.resolve(UUID.randomUUID().toString()); Files.write(parquetTmpFilePath, files.next()); parquetReader = ProtoParquetReader.<SdiiMessage.Message.Builder>builder( new org.apache.hadoop.fs.Path(parquetTmpFilePath.toString())) .build(); for (SdiiMessage.Message.Builder sdiiMessageBuilder; (sdiiMessageBuilder = parquetReader.read()) != null; ) { parquetWriter.write(sdiiMessageBuilder.build()); } parquetReader.close(); deleteTempParquetFile(parquetTmpFilePath); } parquetWriter.close(); return Files.readAllBytes(parquetTargetFilePath); } catch (IOException e) { LOG.error("Error merging files (key={})", keys.values(), e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; }  @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files); }
ParquetCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { Path parquetTmpFilePath = null; Path parquetTargetFilePath = null; try { parquetTargetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetReader<SdiiMessage.Message.Builder> parquetReader; ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetTargetFilePath); while (files.hasNext()) { parquetTmpFilePath = tmpDir.resolve(UUID.randomUUID().toString()); Files.write(parquetTmpFilePath, files.next()); parquetReader = ProtoParquetReader.<SdiiMessage.Message.Builder>builder( new org.apache.hadoop.fs.Path(parquetTmpFilePath.toString())) .build(); for (SdiiMessage.Message.Builder sdiiMessageBuilder; (sdiiMessageBuilder = parquetReader.read()) != null; ) { parquetWriter.write(sdiiMessageBuilder.build()); } parquetReader.close(); deleteTempParquetFile(parquetTmpFilePath); } parquetWriter.close(); return Files.readAllBytes(parquetTargetFilePath); } catch (IOException e) { LOG.error("Error merging files (key={})", keys.values(), e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; }  @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files);  }