Как прочитать все данные Common Crawl из AWS с Java?

Question

Я совершенно новичок в программировании Hadoop и MapReduce, и я пытаюсь написать свою первую программу MapReduce с данными Common Crawl.

Я хотел бы прочитать все данные за апрель 2015 года от AWS. Например, если я хочу загрузить все данные за апрель 2015 года в командной строке, я делаю:

s3cmd get s3: //aws-publicdatasets/common-crawl/crawl-data/CC-MAIN-2015-18/segments/1429246633512.41/wat/*.warc.wat.gz

Эта командная строка работает, но я не хочу загружать все данные за апрель 2015 года, я просто хочу прочитать все файлы "warc.wat.gz" (чтобы проанализировать данные).

Я пытаюсь создать свою работу, которая выглядит так:

public class FirstJob extends Configured implements Tool {
    private static final Logger LOG = Logger.getLogger(FirstJob.class);

    /**
     * Main entry point that uses the {@link ToolRunner} class to run the Hadoop
     * job.
     */
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new FirstJob(), args);
        System.out.println("done !!");
        System.exit(res);
    }

    /**
     * Builds and runs the Hadoop job.
     * 
     * @return 0 if the Hadoop job completes successfully and 1 otherwise.
     */
    public int run(String[] arg0) throws Exception {
        Configuration conf = getConf();
        //
        Job job = new Job(conf);
        job.setJarByClass(FirstJob.class);
        job.setNumReduceTasks(1);

        //String inputPath = "data/*.warc.wat.gz";
        String inputPath = "s3n://aws-publicdatasets/common-crawl/crawl-data/CC-MAIN-2015-18/segments/1429246633512.41/wat/*.warc.wat.gz";
        LOG.info("Input path: " + inputPath);
        FileInputFormat.addInputPath(job, new Path(inputPath));

        String outputPath = "/tmp/cc-firstjob/";
        FileSystem fs = FileSystem.newInstance(conf);
        if (fs.exists(new Path(outputPath))) {
            fs.delete(new Path(outputPath), true);
        }
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        job.setInputFormatClass(WARCFileInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        job.setMapperClass(FirstJobUrlTypeMap.ServerMapper.class);
        job.setReducerClass(LongSumReducer.class);

        if (job.waitForCompletion(true)) {
            return 0;
        } else {
            return 1;
        }
    }

Но у меня есть эта ошибка:

Исключение в потоке "main" java.lang.IllegalArgumentException: идентификатор ключа доступа AWS и секретный ключ доступа должны быть указаны в качестве имени пользователя или пароля (соответственно) для URL-адреса s3n или путем установки fs.s3n.awsAccess KeyId или fs.s3n. Свойства awsSecretAccess Key (соответственно).

Как я могу решить мою проблему? заранее спасибо,

Answer 1

Вы можете попробовать этот проект GitHub.

Answer 2

Я решил свою проблему. В коде измените:

 Configuration conf = getConf();
 //
 Job job = new Job(conf);

в

Configuration conf = new Configuration();
conf.set("fs.s3n.awsAccessKeyId", "your_key");
conf.set("fs.s3n.awsSecretAccessKey", "your_key");
Job job = new Job(conf);