Hadoop - пример конфигурации (настройки) Job с несколькими разными входными путями и несколькими специальными мэпперами

Пример добавление в конфигурацию задачи нескольких входных путей (к папкам и конкретным файлам)

Job job = new Job(conf, "Text about Job =) ");
			job.setJarByClass(driver.class);
		//	job.setMapperClass(Map2_2.class); // мэппер мы укажем ниже
		//	job.setCombinerClass(Red2_2.class);
			job.setReducerClass(Red2_3.class); // указываем класс редуктора
			
			//job.setNumReduceTasks(0); // zero reduce (можно и без редукторов)
			//job.setInputFormatClass(SequenceFileInputFormat.class); // может быть и ПРОсто тексТ
			
			// двумя строчками добавляем справочник в распределённую память
			//URI someFile = new URI("/user/somefile.csv");
			//DistributedCache.addCacheFile(someFile, job.getConfiguration());
			
			/* в четырёх строчка конфигурируем форматы ключа и значения
                            для мэппера и редуктора (ключ-значение для кажого - начинаем с мэппера)*/
                        job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(Text.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(NullWritable.class);
			
			
			//FileSystem hdfs = FileSystem.get(conf);

                        // добавим путь к первой директории (она в нашем пример будет одна) 
                        // - хотя можно добавить сколько угодно отдельных директорий и поддиректорий.
			MultipleInputs.addInputPath(job, new Path("/path/to/yourdir1"),
					TextInputFormat.class, Map2_3.class);// мэппер как мы видим указывается тут же
			// почитаем ещё этот файл
			MultipleInputs.addInputPath(job, new Path("/path/to/yourdir23/somefile1.txt"),
					TextInputFormat.class, Map2_3_ref_1.class); // мэппер для файла 
             
			// и этот файл тоже подадим на вход задачи
			MultipleInputs.addInputPath(job, new Path("/path/to/yourdir255/somefile55.csv"),
					TextInputFormat.class, Map2_3_ref_2.class); // другой класс мэппера для файла

			
			//FileInputFormat.addInputPath(job, new Path(args[0]));
			FileOutputFormat.setOutputPath(job, new Path(args[1])); // напр. получив путь к выходной из командной строки
			
			System.exit(job.waitForCompletion(true) ? 0 : 1);