@Override
    public void prepareToRead(final RecordReader recordReader, final PigSplit pigSplit) throws IOException {
        try {

            int j = 0;
            for(String s : pigSplit.getLocations()) {
                System.out.println((j++) + "Split : " + s);
            }
            System.out.println("Num paths: " + pigSplit.getNumPaths());
            System.out.println("" + pigSplit.getConf());
            System.out.println("split index " + pigSplit.getSplitIndex());


            Thread progressThread = new Thread(new Runnable() {
                public void run() {
                    int i = 0;
                    while(!ready) {
                        PigStatusReporter.getInstance().progress();
                        PigStatusReporter.getInstance().setStatus("GraphChi running (" + i + "): " + getStatusString());
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException ioe) {}
                    }

                }

            });
            progressThread.start();

            if (pigSplit.getSplitIndex() > 0) {
                PigStatusReporter.getInstance().setStatus("Redundant GraphChi-mapper - will die");
                throw new RuntimeException("Split index > 0 -- this mapper will die (expected, not an error).");
            }

            activeNode = true;

            Thread chiThread = new Thread(new Runnable() {
                public void run() {
                    try {
                        setStatusString("Preprocessing: reading data from HDFS: " + location);
                        final FastSharder sharder = createSharder(getGraphName(), getNumShards());


                        HDFSGraphLoader hdfsLoader = new HDFSGraphLoader(location, new EdgeProcessor<Float>() {
                            long counter = 0;

                            public Float receiveEdge(int from, int to, String token) {
                                try {
                                    sharder.addEdge(from, to, token);
                                    counter++;
                                    if (counter % 100000 == 0) {
                                        setStatusString("Preprocessing, read " + counter + " edges");
                                    }
                                } catch (IOException e) {
                                    throw new RuntimeException(e);
                                }
                                return null;
                            }
                        });

                        hdfsLoader.load(pigSplit.getConf());

                        setStatusString("Sharding...");
                        sharder.process();

                        logger.info("Starting to run GraphChi");
                        setStatusString("Start GraphChi engine");
                        runGraphChi();
                        logger.info("Ready.");
                    } catch (Exception err) {
                        err.printStackTrace();
                    }
                    ready = true;
                }});
            chiThread.start();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }