D-Net

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

import java.io.IOException;

import java.io.StringReader;

import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.dom4j.Document;

import org.dom4j.io.SAXReader;

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

	@Override

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

		logConfiguration(context.getConfiguration());

	}

	@Override

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

		try {

			final SAXReader reader = new SAXReader();

			final Document doc = reader.read(new StringReader(value.toString()));

			final String id = doc.valueOf("//dri:objIdentifier");

			addDocument(context, id, value.toString());

			context.getCounter("Document Database", "records").increment(1);

		} catch (Exception e) {

			context.getCounter("Document Database", "errors").increment(1);

			context.write(key, value);

			e.printStackTrace(System.err);

		}

	}

	private void addDocument(final Context context, final String id, final String string) {

		// TODO

		// ADD the document to the DB

	}

	@Override

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

		super.cleanup(context);

	}

	private void logConfiguration(final Configuration conf) {

		System.out.println("job configutation #################");

		for (Entry<String, String> e : conf) {

			System.out.println("'" + e.getKey() + "' : '" + e.getValue() + "'");

		}

		System.out.println("end of job configutation #################\n\n");

	}

}