Project

General

Profile

1
package eu.dnetlib.data.mapreduce.hbase.propagation.projecttoresult;
2

    
3

    
4
import com.googlecode.protobuf.format.JsonFormat;
5
import eu.dnetlib.data.mapreduce.hbase.propagation.NotValidResultSequenceException;
6
import eu.dnetlib.data.mapreduce.hbase.propagation.ResultIterator;
7
import eu.dnetlib.data.proto.*;
8
import org.apache.commons.logging.Log;
9
import org.apache.commons.logging.LogFactory;
10
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
11
import org.apache.hadoop.hbase.util.Bytes;
12
import org.apache.hadoop.io.Text;
13
import org.apache.hadoop.mapreduce.Reducer;
14

    
15
import java.io.IOException;
16

    
17
import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;
18

    
19
public class PropagationProjectToResultFileReducer extends Reducer<ImmutableBytesWritable, Text, Text,Text> {
20
    private static final Log log = LogFactory.getLog(PropagationProjectToResultReducer.class); // NOPMD by marko on 11/24/08 5:02 PM
21

    
22
    private Text keyOut;
23
    private Text outValue;
24

    
25

    
26
    @Override
27
    protected void setup(final Context context) throws IOException, InterruptedException {
28
        super.setup(context);
29
        keyOut = new Text("");
30
        outValue = new Text();
31
    }
32

    
33

    
34
    @Override
35
    protected void reduce(ImmutableBytesWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
36
        ResultIterator rh = null;
37
        try {
38
            rh = new ResultProjectIterator(values, Bytes.toString(key.copyBytes()));
39
        } catch (NotValidResultSequenceException e) {
40
            context.getCounter(COUNTER_PROPAGATION, e.getMessage()).increment(1);
41
            return;
42
        }
43
        while (rh.hasNext()) {
44
            for (OafProtos.Oaf oaf : rh.next()) {
45
                keyOut.set(oaf.getRel().getTarget());
46
                outValue.set(JsonFormat.printToString(oaf).getBytes());
47
                context.write(keyOut, outValue);
48
            }
49
            context.getCounter(COUNTER_PROPAGATION, "Added relation to project").increment(1);
50
        }
51

    
52
    }
53
}
(2-2/5)