Project

General

Profile

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

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

    
16
import java.io.IOException;
17
import java.util.List;
18

    
19
import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.COUNTER_PROPAGATION;
20

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

    
24
    private Text keyOut;
25
    private Text outValue;
26

    
27

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

    
35

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

    
55
            }
56

    
57
        }
58

    
59
    }
60
}
(2-2/5)