Блог

Я создал подкласс под названием TextArrayWritable для хранения массива текста. Это выглядит так:

 import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.Text;
public class TextArrayWritable extends ArrayWritable{
    public TextArrayWritable() {
        super(Text.class);
    }
    public TextArrayWritable(Text[] values) {
        super(Text.class,values);
    }
}
  

В моем картографе я создаю пару ключ-значение, где ключом является текст, а значением является приведенный выше TextArrayWritable. Код для картографа выглядит так, как показано ниже:

 import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


public class TagWordMapper extends Mapper<LongWritable,Text,Text,TextArrayWritable>{
    
    @Override
    public void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException{
        if(key.get()==0) {
            return;
        }else {
            String line = value.toString();
            Text title = new Text(line.split("t")[2]);
            Text likes =  new Text(line.split("t")[8]);
            Text tags =  new Text(line.split("t")[6]);
            
            Text[] temp = new Text[2];
            temp[0]=likes;
            temp[1]=tags;
            
            context.write(title, new TextArrayWritable(temp));
        }
    }
}
  

Проблема возникает в редукторе, когда я перебираю TextArrayWritable, возникает исключение, даже если я привел его как Text[], поскольку TextArrayWritable#get() возвращает массив текста. Код редуктора выглядит следующим образом:

 import java.io.IOException;
import java.util.regex.*;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class TagWordCombiner extends Reducer<Text,TextArrayWritable,Text,IntWritable>{

    @Override
    public void reduce(Text title,Iterable<TextArrayWritable> arrays,Context context) throws IOException, InterruptedException{
        int count=0;
        Pattern r = Pattern.compile("\bcute\b",Pattern.CASE_INSENSITIVE);
        for(TextArrayWritable array:arrays) {
            
            Text[] temp = (Text[]) array.get();
            int likes = Integer.parseInt(temp[0].toString());
            String tags = temp[1].toString();
            
            Matcher matcher = r.matcher(tags);
            boolean found = matcher.find();
            if(found amp;amp; likes>=3000){
                count =1;
            }
        }
        context.write(title, new IntWritable(count));
    }
}
  

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

Я решил не использовать TextArrayWritable и просто решил отправлять массивы данных в строковой форме с использованием текста. Более того, я обнаружил, что для Комбайнеров это работает. Тип данных ввода / вывода объединителя должен соответствовать типу выходных данных преобразователя.

Вот мой код:

Mapper:

 package combiner;
import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class CombinerMapper extends Mapper<LongWritable,Text,Text,Text>{

    @Override
    public void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{
        if(key.get()==0) {
            return;
        }else {
            String line = value.toString();
            Text title = new Text(line.split("t")[2]);
            Text arr = new Text(line.split("t")[8] "t" line.split("t")[6]);
            context.write(title, arr);
        }
        
    }
}
  

Что он делает, он отображает .csv файл. Он пропускает первую строку, потому что первая строка является заголовками и бесполезна. В следующей строке он разделяется на » t», чтобы извлечь «заголовок», «лайки», «теги» в таком порядке. В массив должны быть вставлены только «лайки» и «теги». Тип выходных данных Mapper — текст,Text.

Объединитель:

 package combiner;
import java.util.regex.*;
import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class CombinerCombiner extends Reducer<Text,Text,Text,Text>{

    @Override
    public void reduce(Text title,Iterable<Text> values,Context context) throws IOException, InterruptedException{
        int sum=0;
        for(Text val:values) {
            String line = val.toString();
            int likes = Integer.parseInt(line.split("t")[0]);
            String tags = line.split("t")[1];
            
            Pattern r = Pattern.compile("\bcute\b",Pattern.CASE_INSENSITIVE);
            Matcher matcher = r.matcher(tags);
            boolean found = matcher.find();
            
            if(found amp;amp; likes>=3000) {
                sum =1;
            }
        }
        if(sum==0) {
            return;
        }else {
            context.write(title, new Text(Integer.toString(sum)));
        }
    }
}
  

Сложность в том, как работает объединитель, заключается в том, что его ввод / вывод должен соответствовать типу выходных данных Mapper, иначе он выдаст ошибку, с которой я столкнулся. В этом случае я использую объединитель в качестве фильтра, чтобы отфильтровывать заголовки, которые не имеют >= 3000 лайков, а его теги не содержат слова cute ни в какой форме. Он вводит как <Текст,Text> тип данных и выводит как <Текст,Text> тип данных, оба совпадают с типом выходных данных Mapper.

Наконец, редуктор:

 package combiner;
import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class CombinerReducer extends Reducer<Text,Text,Text,IntWritable>{

    public void reduce(Text title,Iterable<Text>counts,Context context) throws IOException, InterruptedException{
        int sum = 0;
        for(Text count:counts) {
            String line = count.toString();
            int temp = Integer.parseInt(line);
            sum =temp;
        }
        context.write(title, new IntWritable(sum));
    }
  

Он функционирует как обычный редуктор, он суммирует массивы данных и выдает конечную сумму. Перебирать итерацию совершенно бессмысленно, поскольку это делается для нас в объединителе, но это работает.

Вот мой класс драйвера:

 package combiner;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.mapreduce.Job;

public class Combiner {

    public static void main(String[] args) throws Exception{
        //Check if proper arguments is inputed
        if (args.length != 2) {
              System.out.printf(
                  "Usage: TagWord <input dir> <output dir>n");
              System.exit(-1);
        }
        
        Job job = new Job();
        job.setJarByClass(Combiner.class);
        job.setJobName("Combiner");
        
        //Declare the input file path
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        
        //Declare the output file path
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        //Initiate the the -cers
        job.setMapperClass(CombinerMapper.class);
        job.setCombinerClass(CombinerCombiner.class);
        job.setReducerClass(CombinerReducer.class);
        
        //map output
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        
        //reducer output
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        //Initiate boolean to check whether MapReduce is successful
        boolean success = job.waitForCompletion(true);
        
        //Execute code based on whether the job is successful
        System.exit(success ? 0 : 1);
    }
}
  

Обратите внимание на класс combiner, расположенный между классом mapper и reducer. Кроме того, убедитесь, что включены setMapOutputKeyClass и setMapOutputValueClass . Это позволяет использовать другой тип выходных данных для класса Mapper. Это также заставляет тип выходных данных редуктора быть таким, какой вы установили для setOutputKeyClass и setOutputValueClass, в нашем случае наш редуктор должен выводить текст и быть записываемым.

Если есть какие-либо вопросы, прокомментируйте. Информации для Hadoop немного мало.