D-Net

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

import java.io.IOException;

import java.util.Iterator;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

    private static final Log log = LogFactory.getLog(PropagationCountryInstitutionalOrganizationReducer.class);

    private ImmutableBytesWritable keyOut;

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

        super.setup(context);

    @Override

        if(!it.hasNext()){

        }

			throw new RuntimeException("First Element in reducer is not type of datasource");

		}

            context.getCounter(COUNTER_PROPAGATION,"institutional datasource").increment(1);

            }

            final String country = Bytes.toString(it.next().get()); // need to update the information for the country to each element in the iterator

            }

			while(it.hasNext()) {

                byte[] targetRowKey = it.next().get();

				String resultId = Bytes.toString(targetRowKey);

				if (!resultId.startsWith(("50|"))) {

					context.getCounter(COUNTER_PROPAGATION,"resultId expected in ordering was not found").increment(1);

                    //throw new RuntimeException("Problem in ordering. Elements different from publication in list after second position " + targetRowKey);

                final Put put = new Put(targetRowKey).add(Bytes.toBytes("result"), Bytes.toBytes("update_"+System.nanoTime()), oafUpdate);

                context.write(keyOut, put);

                context.getCounter(COUNTER_PROPAGATION, " added country to product ").increment(1);

            }

        } else {

            context.getCounter(COUNTER_PROPAGATION,"not institutional datasource").increment(1);

        }

			.setClassid(countryValue)

			.setClassname(countryValue)

			.setSchemeid(DNETCOUNTRYSCHEMA)

			.setSchemename(DNETCOUNTRYSCHEMA);

		final ResultProtos.Result.Builder result = ResultProtos.Result.newBuilder().setMetadata(metadata);

		final OafProtos.OafEntity.Builder entity = OafProtos.OafEntity.newBuilder()

				.setType(TypeProtos.Type.result)

				.setId(resultId)

				.setResult(result);

				.setKind(KindProtos.Kind.entity)

				.setEntity(entity)

				.build()

				.toByteArray();

	}

}