ParquetStorer extends StoreFunc implements StoreMetadata { private Schema getSchema() { try { return Utils.getSchemaFromString(getProperties().getProperty(SCHEMA)); } catch (ParserException e) { throw new RuntimeException("can not get schema from context", e); } } ParquetStorer(); @Override void setStoreFuncUDFContextSignature(String signature); @Override void checkSchema(ResourceSchema s); @Override OutputFormat<Void, Tuple> getOutputFormat(); @SuppressWarnings({ "rawtypes", "unchecked" }) // that's how the base class is defined @Override void prepareToWrite(RecordWriter recordWriter); @Override void putNext(Tuple tuple); @Override void setStoreLocation(String location, Job job); @Override void storeSchema(ResourceSchema schema, String location, Job job); @Override void storeStatistics(ResourceStatistics resourceStatistics, String location, Job job);  }
@Test public void testComplexSchema() throws ExecException, Exception { String out = "target/out"; PigServer pigServer = new PigServer(ExecType.LOCAL); Data data = Storage.resetData(pigServer); Collection<Tuple> list = new ArrayList<Tuple>(); for (int i = 0; i < 1000; i++) { list.add(tuple("a"+i, bag(tuple("o", "b")))); } for (int i = 10; i < 2000; i++) { list.add(tuple("a"+i, bag(tuple("o", "b"), tuple("o", "b"), tuple("o", "b"), tuple("o", "b")))); } for (int i = 20; i < 3000; i++) { list.add(tuple("a"+i, bag(tuple("o", "b"), tuple("o", null), tuple(null, "b"), tuple(null, null)))); } for (int i = 30; i < 4000; i++) { list.add(tuple("a"+i, null)); } Collections.shuffle((List<?>)list); data.set("in", "a:chararray, b:{t:(c:chararray, d:chararray)}", list ); pigServer.setBatchOn(); pigServer.registerQuery("A = LOAD 'in' USING mock.Storage();"); pigServer.deleteFile(out); pigServer.registerQuery("Store A into '"+out+"' using "+ParquetStorer.class.getName()+"();"); if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) { throw new RuntimeException("Job failed", pigServer.executeBatch().get(0).getException()); } { pigServer.registerQuery("B = LOAD '"+out+"' USING "+ParquetLoader.class.getName()+"();"); pigServer.registerQuery("Store B into 'out' using mock.Storage();"); if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) { throw new RuntimeException("Job failed", pigServer.executeBatch().get(0).getException()); } List<Tuple> result = data.get("out"); assertEquals(list, result); final Schema schema = data.getSchema("out"); assertEquals("{a:chararray, b:{t:(c:chararray, d:chararray)}}".replaceAll(" ", ""), schema.toString().replaceAll(" ", "")); } { pigServer.registerQuery("C = LOAD '"+out+"' USING "+ParquetLoader.class.getName()+"('a:chararray');"); pigServer.registerQuery("Store C into 'out2' using mock.Storage();"); if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) { throw new RuntimeException("Job failed", pigServer.executeBatch().get(0).getException()); } final Function<Tuple,Object> grabFirstColumn = new Function<Tuple,Object>() { @Override public Object apply(Tuple input) { try { return input.get(0); } catch (ExecException e) { throw new RuntimeException(e); } } }; List<Tuple> result2 = data.get("out2"); Object[] result2int = Collections2.transform(result2, grabFirstColumn).toArray(); Object[] input2int = Collections2.transform(list, grabFirstColumn).toArray(); assertArrayEquals(input2int, result2int); } }