@Test public void testMerge() throws IOException { 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); Path tmpDir = Files.createTempDirectory("parquetTmp"); tmpDir.toFile().deleteOnExit(); Path parquetTmpFilePath = tmpDir.resolve(UUID.randomUUID().toString()); Files.write(parquetTmpFilePath, mergedByteArray); 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(m3, reader.read().build()); assertEquals(m4, reader.read().build()); assertEquals(null, reader.read()); reader.close(); parquetTmpFilePath.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("Errors merging files....", e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; }
ParquetSimpleKeyExample implements SimpleUDF { @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("Errors merging files....", e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; } }
ParquetSimpleKeyExample implements SimpleUDF { @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("Errors merging files....", e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } return null; }  }
ParquetSimpleKeyExample implements SimpleUDF { @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("Errors merging files....", e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } 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[] 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("Errors merging files....", e); } finally { deleteTempParquetFile(parquetTmpFilePath); deleteTempParquetFile(parquetTargetFilePath); } 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);  }