Method testMultipleMRJobs

test/org/apache/pig/test/TestCounters.java:400–462

()

    }

    @Test
    public void testMultipleMRJobs() throws IOException, ExecException {
        Assume.assumeTrue("Skip this test for TEZ. Assert is done only for first MR job",
                Util.isMapredExecType(cluster.getExecType()));
        int count = 0;
        PrintWriter pw = new PrintWriter(Util.createInputFile(cluster, file));
        int [] nos = new int[10];
        for(int i = 0; i < 10; i++)
            nos[i] = 0;

        for(int i = 0; i < MAX; i++) {
            int index = r.nextInt(10);
            int value = r.nextInt(100);
            nos[index] += value;
            pw.println(index + "\t" + value);
        }
        pw.close();

        for(int i = 0; i < 10; i++) {
            if(nos[i] > 0) count ++;
        }

        PigServer pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
        pigServer.registerQuery("a = load '" + file + "';");
        pigServer.registerQuery("b = order a by $0;");
        pigServer.registerQuery("c = group b by $0;");
        pigServer.registerQuery("d = foreach c generate group, SUM(b.$1);");
        ExecJob job = pigServer.store("d", "output");
        PigStats pigStats = job.getStatistics();

        InputStream is = FileLocalizer.open(FileLocalizer.fullPath("output",
                pigServer.getPigContext()), pigServer.getPigContext());
        long filesize = 0;
        while(is.read() != -1) filesize++;

        is.close();

        cluster.getFileSystem().delete(new Path(file), true);
        cluster.getFileSystem().delete(new Path("output"), true);

        System.out.println("============================================");
        System.out.println("Test case MultipleMRJobs");
        System.out.println("============================================");

        JobGraph jp = pigStats.getJobGraph();
        JobStats js = (JobStats)jp.getSinks().get(0);

        System.out.println("Job id: " + js.getName());
        System.out.println(jp.toString());

        System.out.println("Map input records : " + js.getMapInputRecords());
        assertEquals(MAX, js.getMapInputRecords());
        System.out.println("Map output records : " + js.getMapOutputRecords());
        assertEquals(MAX, js.getMapOutputRecords());
        System.out.println("Reduce input records : " + js.getReduceInputRecords());
        assertEquals(count, js.getReduceInputRecords());
        System.out.println("Reduce output records : " + js.getReduceOutputRecords());
        assertEquals(count, js.getReduceOutputRecords());