public void load(Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);

        Path inputPath = new Path(hdfsLocation);

        if (fs.isFile(inputPath)) {
            processFile(fs, inputPath);
        } else {
            FileStatus[] dirFiles = fs.listStatus(inputPath);
            for(FileStatus status : dirFiles) {
                logger.info("Dir entry: " + status.toString() + " " + status.getPath().getName());
                if (status.getPath().getName().startsWith("part-")) {
                    processFile(fs, status.getPath());
                }
            }
        }
    }