Method testBzipInPig

test/org/apache/pig/test/TestBZip.java:97–148

Tests the end-to-end writing and reading of a BZip file.

()

    * Tests the end-to-end writing and reading of a BZip file.
    */
    @Test
    public void testBzipInPig() throws Exception {
        PigServer pig = new PigServer(cluster.getExecType(), properties);

        File in = folder.newFile("junit-in.bz2");

        File out = folder.newFile("junit-out.bz2");
        out.delete();
        String clusterOutput = Util.removeColon(out.getAbsolutePath());

        CBZip2OutputStream cos =
            new CBZip2OutputStream(new FileOutputStream(in));
        for (int i = 1; i < 100; i++) {
            StringBuffer sb = new StringBuffer();
            sb.append(i).append("\n").append(-i).append("\n");
            byte bytes[] = sb.toString().getBytes();
            cos.write(bytes);
        }
        cos.close();

        pig.registerQuery("AA = load '"
                + Util.generateURI(in.getAbsolutePath(), pig.getPigContext())
                + "';");
        pig.registerQuery("A = foreach (group (filter AA by $0 > 0) all) generate flatten($1);");
        pig.registerQuery("store A into '" + Util.encodeEscape(clusterOutput) + "';");
        FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(
                pig.getPigContext().getProperties()));
        FileStatus[] outputFiles = fs.listStatus(new Path(clusterOutput),
                Util.getSuccessMarkerPathFilter());
        FSDataInputStream is = fs.open(outputFiles[0].getPath());
        CBZip2InputStream cis = new CBZip2InputStream(is, -1, out.length());

        // Just a sanity check, to make sure it was a bzip file; we
        // will do the value verification later
        assertEquals(100, cis.read(new byte[100]));
        cis.close();

        pig.registerQuery("B = load '" + Util.encodeEscape(clusterOutput) + "';");

        Iterator<Tuple> i = pig.openIterator("B");
        HashMap<Integer, Integer> map = new HashMap<Integer, Integer>();
        while (i.hasNext()) {
            Integer val = DataType.toInteger(i.next().get(0));
            map.put(val, val);
        }

        assertEquals(new Integer(99), new Integer(map.keySet().size()));

        for (int j = 1; j < 100; j++) {
            assertEquals(new Integer(j), map.get(j));
        }
    }

   /**
    * Tests the end-to-end writing and reading of a BZip file using absolute path with a trailing /.