@Test public void testAggregate() throws IOException { List<byte[]> list = Arrays.asList(b1, b2); Iterator it = list.iterator(); byte[] aggregatedByteArray = example.aggregate(new HashMap<>(), it); Path tmpDir = Files.createTempDirectory("parquetTmp"); tmpDir.toFile().deleteOnExit(); Path parquetTmpFilePath = tmpDir.resolve(UUID.randomUUID().toString()); Files.write(parquetTmpFilePath, aggregatedByteArray); ParquetReader<SdiiMessage.Message.Builder> reader = ProtoParquetReader.<SdiiMessage.Message.Builder>builder( new org.apache.hadoop.fs.Path(parquetTmpFilePath.toString())) .build(); assertEquals(m1, reader.read().build()); assertEquals(m2, reader.read().build()); assertEquals(null, reader.read()); reader.close(); parquetTmpFilePath.toFile().delete(); }
@Override public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) { Path parquetFilePath = null; try { parquetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetFilePath); while (messages.hasNext()) { parquetWriter.write(SdiiMessage.Message.parseFrom(messages.next())); } parquetWriter.close(); return Files.readAllBytes(parquetFilePath); } catch (IOException e) { LOG.error("Errors creating parquet temp file....", e); } finally { deleteTempParquetFile(parquetFilePath); } return null; }
ParquetSimpleKeyExample implements SimpleUDF { @Override public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) { Path parquetFilePath = null; try { parquetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetFilePath); while (messages.hasNext()) { parquetWriter.write(SdiiMessage.Message.parseFrom(messages.next())); } parquetWriter.close(); return Files.readAllBytes(parquetFilePath); } catch (IOException e) { LOG.error("Errors creating parquet temp file....", e); } finally { deleteTempParquetFile(parquetFilePath); } return null; } }
ParquetSimpleKeyExample implements SimpleUDF { @Override public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) { Path parquetFilePath = null; try { parquetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetFilePath); while (messages.hasNext()) { parquetWriter.write(SdiiMessage.Message.parseFrom(messages.next())); } parquetWriter.close(); return Files.readAllBytes(parquetFilePath); } catch (IOException e) { LOG.error("Errors creating parquet temp file....", e); } finally { deleteTempParquetFile(parquetFilePath); } return null; }  }
ParquetSimpleKeyExample implements SimpleUDF { @Override public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) { Path parquetFilePath = null; try { parquetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetFilePath); while (messages.hasNext()) { parquetWriter.write(SdiiMessage.Message.parseFrom(messages.next())); } parquetWriter.close(); return Files.readAllBytes(parquetFilePath); } catch (IOException e) { LOG.error("Errors creating parquet temp file....", e); } finally { deleteTempParquetFile(parquetFilePath); } return null; }  @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); }
ParquetSimpleKeyExample implements SimpleUDF { @Override public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) { Path parquetFilePath = null; try { parquetFilePath = tmpDir.resolve(UUID.randomUUID().toString()); ParquetWriter<SdiiMessage.Message> parquetWriter = createParquetWriter(parquetFilePath); while (messages.hasNext()) { parquetWriter.write(SdiiMessage.Message.parseFrom(messages.next())); } parquetWriter.close(); return Files.readAllBytes(parquetFilePath); } catch (IOException e) { LOG.error("Errors creating parquet temp file....", e); } finally { deleteTempParquetFile(parquetFilePath); } return null; }  @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);  }