D-Net

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

import java.io.IOException;

import java.util.List;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import com.google.common.collect.Lists;

import eu.dnetlib.miscutils.datetime.HumanTime;

public class StatsFeedMapper extends Mapper<Text, Text, Text, Text> {

	private List<String> buffer;

	private int bufferFlushThreshold = 1000;

	@Override

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

		buffer = Lists.newArrayList();

	}

	@Override

	protected void map(final Text key, final Text value, final Context context) throws IOException, InterruptedException {

		final String copyCmd = buildCopyCommand(value);

		try {

			if ((copyCmd != null) && !copyCmd.isEmpty()) {

				buffer.add(copyCmd);

				if (buffer.size() >= bufferFlushThreshold) {

					doAdd(buffer, context);

				}

			} else {

				context.getCounter("stats", "skipped records").increment(1);

			}

		} catch (Throwable e) {

			context.getCounter("stats", e.getClass().toString()).increment(1);

			context.write(key, printRottenRecord(context.getTaskAttemptID().toString(), value, copyCmd));

			e.printStackTrace(System.err);

		}

	}

	private String buildCopyCommand(final Text value) {

		return null;

	}

	private void doAdd(final List<String> buffer, final Context context) {

		long start = System.currentTimeMillis();

		// flush the buffer

		// UpdateResponse rsp = serverPool.addAll(buffer);

		long stop = System.currentTimeMillis() - start;

		System.out.println("feed time for " + buffer.size() + " records : " + HumanTime.exactly(stop) + "\n");

		buffer.clear();

	}

	@Override

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

		super.cleanup(context);

		if (!buffer.isEmpty()) {

			doAdd(buffer, context);

		}

	}

	private Text printRottenRecord(final String taskid, final Text value, final String copyCmq) {

		return new Text("\n**********************************\n" + "task: " + taskid + "\n" + check("original", value.toString() + check("copy cmd", copyCmq)));

	}

	private String check(final String label, final Object value) {

		if ((value != null) && !value.toString().isEmpty()) { return "\n " + label + ":\n" + value + "\n"; }

		return "\n";

	}

}