D-Net

                if(StringUtils.isNotBlank(v.getValue()))

                    fromResult.addAll(Arrays.asList(StringUtils.split(v.getValue(), ",")));

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

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.mapreduce.hbase.propagation.NotValidResultSequenceException;

import eu.dnetlib.data.mapreduce.hbase.propagation.ResultIterator;

import eu.dnetlib.data.proto.*;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

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

public class PropagationProjectToResultFileReducer extends Reducer<ImmutableBytesWritable, Text, Text,Text> {

    private static final Log log = LogFactory.getLog(PropagationProjectToResultReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

    private Text keyOut;

    private Text outValue;

    @Override

    protected void setup(final Context context) throws IOException, InterruptedException {

        super.setup(context);

        keyOut = new Text("");

        outValue = new Text();

    }

    @Override

    protected void reduce(ImmutableBytesWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        ResultIterator rh = null;

        try {

            rh = new ResultProjectIterator(values, Bytes.toString(key.copyBytes()));

        } catch (NotValidResultSequenceException e) {

            context.getCounter(COUNTER_PROPAGATION, e.getMessage()).increment(1);

            return;

        }

        while (rh.hasNext()) {

            for (OafProtos.Oaf oaf : rh.next()) {

                keyOut.set(oaf.getRel().getTarget());

                outValue.set(JsonFormat.printToString(oaf).getBytes());

                context.write(keyOut, outValue);

            }

            context.getCounter(COUNTER_PROPAGATION, "Added relation to project").increment(1);

        }

    }

}

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

import eu.dnetlib.data.mapreduce.hbase.propagation.NotValidResultSequenceException;

import eu.dnetlib.data.mapreduce.hbase.propagation.ResultIterator;

import eu.dnetlib.data.proto.*;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import java.io.IOException;

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

public class PropagationProjectToResultReducer extends TableReducer<ImmutableBytesWritable, Text, ImmutableBytesWritable> {

    private static final Log log = LogFactory.getLog(PropagationProjectToResultReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

    private ImmutableBytesWritable keyOut;

    @Override

    protected void setup(final Context context) throws IOException, InterruptedException {

        super.setup(context);

        keyOut = new ImmutableBytesWritable();

    }

    @Override

    protected void reduce(ImmutableBytesWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        ResultIterator rh = null;

        try {

            rh = new ResultProjectIterator(values,Bytes.toString(key.copyBytes()));

        } catch (NotValidResultSequenceException e) {

            context.getCounter(COUNTER_PROPAGATION,e.getMessage()).increment(1);

            return;

        }

        while(rh.hasNext()){

            for(OafProtos.Oaf oaf:rh.next()){

                final String source = oaf.getRel().getSource();

                final Put put = new Put(Bytes.toBytes(source)).add(Bytes.toBytes(RELATION + oaf.getRel().getRelClass()),Bytes.toBytes(oaf.getRel().getTarget()),oaf.toByteArray());

                keyOut.set(Bytes.toBytes(source));

                context.write(keyOut, put);

            }

            context.getCounter(COUNTER_PROPAGATION,"Added relation to project").increment(1);

        }

    }

}