Как указать KeyValueTextInputFormat Separator в Hadoop-.20 api?

В новом API (apache.hadoop.mapreduce.KeyValueTextInputFormat), как указать разделитель (разделитель), отличный от табуляции (по умолчанию) для разделения ключа и значения.

Пример ввода:

one,first line
two,second line

Обязательный параметр:

Key : one
Value : first line
Key : two
Value : second line

Я указываю KeyValueTextInputFormat как:

    Job job = new Job(conf, "Sample");

    job.setInputFormatClass(KeyValueTextInputFormat.class);
    KeyValueTextInputFormat.addInputPath(job, new Path("/home/input.txt"));

Это отлично работает для вкладок в качестве разделителя.

Ответ 1

В новом API вы должны использовать свойство конфигурации mapreduce.input.keyvaluelinerecordreader.key.value.separator.

Вот пример:

Configuration conf = new Configuration();
conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");

Job job = new Job(conf);
job.setInputFormatClass(KeyValueTextInputFormat.class);
// next job set-up

Ответ 2

В коде драйвера укажите следующее.

conf.set("key.value.separator.in.input.line", ",");

Ответ 3

Для KeyValueTextInputFormat входная строка должна представлять собой пару ключевых значений, разделенную "\ t"

Key1     Value1,Value2

Изменяя по умолчанию разделитель, вы сможете читать, как хотите.

Для нового Api

Вот решение

//New API
Configuration conf = new Configuration();
conf.set("key.value.separator.in.input.line", ","); 
Job job = new Job(conf);
job.setInputFormatClass(KeyValueTextInputFormat.class);

Карта

public class Map extends Mapper<Text, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(Text key, Text value, Context context)
        throws IOException, InterruptedException {
    String line = value.toString();
    System.out.println("key---> "+key);
    System.out.println("value---> "+value.toString());
   .
   .

Выход

key---> one
value---> first line
key---> two
value---> second line

Ответ 4

Это вопрос последовательности.

Первая строка conf.set("key.value.separator.in.input.line", ",") должна появиться перед созданием экземпляра класса Job. Итак:

conf.set("key.value.separator.in.input.line", ","); 
Job job = new Job(conf);

Ответ 5

Во-первых, новый API не закончил в 0.20. * поэтому, если вы хотите использовать новый API в 0.20. *, вы должны реализовать эту функцию самостоятельно. Например, вы можете использовать FileInputFormat для достижения. Игнорируйте ключ LongWritable и разделите значение Text на запятую самостоятельно.

Ответ 6

По умолчанию класс KeyValueTextInputFormat использует вкладку в качестве разделителя для ключа и значения из входного текстового файла.

Если вы хотите прочитать ввод из пользовательского разделителя, вам нужно настроить конфигурацию с помощью используемого вами атрибута.

Для новых API Hadoop это другое:

conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ";");

Ответ 7

пример

public class KeyValueTextInput extends Configured implements Tool {
    public static void main(String args[]) throws Exception {
        String log4jConfPath = "log4j.properties";
        PropertyConfigurator.configure(log4jConfPath);
        int res = ToolRunner.run(new KeyValueTextInput(), args);
        System.exit(res);
    }

    public int run(String[] args) throws Exception {
        **Configuration conf = this.getConf();**
        //conf.set("key.value.separator.in.input.line", ",");
        **conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");**

        Job job = Job.getInstance(conf, "WordCountSampleTemplate");
        job.setJarByClass(KeyValueTextInput.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        //job.setMapOutputKeyClass(Text.class);
        //job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setInputFormatClass(KeyValueTextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        Path outputPath = new Path(args[1]);
        FileSystem fs = FileSystem.get(new URI(outputPath.toString()), conf);
        fs.delete(outputPath, true);
        FileOutputFormat.setOutputPath(job, outputPath);
        return job.waitForCompletion(true) ? 0 : 1;
    }
}

class Map extends Mapper<Text, Text, Text, Text> {
    public void map(Text k1, Text v1, Context context) throws IOException, InterruptedException {
        context.write(k1, v1);
    }
}

class Reduce extends Reducer<Text, Text, Text, Text> {
    public void reduce(Text Key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        String sum = " || ";
        for (Text value : values)
            sum = sum + value.toString() + " || ";
        context.write(Key, new Text(sum));
    }
}