D-Net

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet-parent</artifactId>

		<version>1.0.0</version>

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-hadoop-service</artifactId>

	<packaging>jar</packaging>

	<version>2.4.1</version>

	<scm>

	    <developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.4.1</developerConnection>

	</scm>

	<repositories>

		<!-- Cloudera Repositories -->

		<repository>

			<snapshots>

				<enabled>false</enabled>

			</snapshots>

			<id>cloudera-central</id>

			<name>cloundera-libs-release</name>

			<url>http://maven.research-infrastructures.eu/nexus/content/repositories/cloudera-central</url>

		</repository>

		<repository>

			<id>cloudera-snapshots</id>

			<name>cloudera-libs-snapshot</name>

			<url>http://maven.research-infrastructures.eu/nexus/content/repositories/cloudera-snapshots</url>

		</repository>

	</repositories>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-hadoop-service-rmi</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>com.google.code.gson</groupId>

			<artifactId>gson</artifactId>

			<version>${google.gson.version}</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-blackboard-common</artifactId>

			<version>[2.1.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-resultset-client</artifactId>

			<version>[2.0.0,3.0.0)</version>

			<exclusions>

				<exclusion>

					<artifactId>slf4j-api</artifactId>

					<groupId>org.slf4j</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-mapreduce-submitter</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>org.apache.hbase</groupId>

			<artifactId>hbase</artifactId>

			<version>${apache.hbase.version}</version>

			<exclusions>

				<exclusion>

					<artifactId>jsp-api-2.1</artifactId>

					<groupId>org.mortbay.jetty</groupId>

				</exclusion>

				<exclusion>

					<artifactId>servlet-api-2.5</artifactId>

					<groupId>org.mortbay.jetty</groupId>

				</exclusion>

				<exclusion>

					<artifactId>slf4j-api</artifactId>

					<groupId>org.slf4j</groupId>

				</exclusion>

				<exclusion>

					<artifactId>slf4j-log4j12</artifactId>

					<groupId>org.slf4j</groupId>

				</exclusion>

				<exclusion>

					<artifactId>stax-api</artifactId>

					<groupId>stax</groupId>

				</exclusion>

				<exclusion>

					<artifactId>httpclient</artifactId>

					<groupId>org.apache.httpcomponents</groupId>

				</exclusion>

				<exclusion>

					<artifactId>guava</artifactId>

					<groupId>com.google.guava</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>org.apache.oozie</groupId>

			<artifactId>oozie-client</artifactId>

			<version>${apache.oozie.version}</version>

			<exclusions>

				<exclusion>

					<artifactId>slf4j-simple</artifactId>

					<groupId>org.slf4j</groupId>

				</exclusion>

				<exclusion>

					<artifactId>slf4j-api</artifactId>

					<groupId>org.slf4j</groupId>

				</exclusion>

				<exclusion>

					<artifactId>xercesImpl</artifactId>

					<groupId>xerces</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.logging.log4j</groupId>

			<artifactId>log4j-slf4j-impl</artifactId>

			<version>2.0.2</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.httpcomponents</groupId>

			<artifactId>httpclient</artifactId>

			<version>4.3.1</version>

		</dependency>		

	</dependencies>

</project>

package eu.dnetlib.data.hadoop;

import java.io.IOException;

import java.net.URI;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.NavigableMap;

import java.util.Set;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;

import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;

import eu.dnetlib.data.hadoop.rmi.hbase.Column;

import eu.dnetlib.data.hadoop.rmi.hbase.HBaseRowDescriptor;

import eu.dnetlib.data.hadoop.utils.HadoopUtils;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTableDescriptor;

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

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

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

public class HadoopServiceCore {

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

	@Autowired

	protected ConfigurationEnumerator configurationEnumerator;

	@Autowired

	private HadoopClientMap clients;

	private int maxVersions;

	public List<String> listTables(final ClusterName clusterName) throws IOException, HadoopServiceException {

		final HBaseAdmin admin = getHBaseAdmin(clusterName);

		return Lists.newArrayList(Iterables.transform(Lists.newArrayList(admin.listTables()), HadoopUtils.tableName()));

	}

	private HBaseAdmin getHBaseAdmin(final ClusterName clusterName) throws HadoopServiceException {

		final HBaseAdmin admin = clients.getHbaseAdmin(clusterName);

		if (admin == null) throw new HadoopServiceException(String.format("HBase admin not available for cluster: '%s'", clusterName.toString()));

		return admin;

	}

	public List<String> describeTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {

		final HBaseAdmin admin = getHBaseAdmin(clusterName);

		final HTableDescriptor desc = admin.getTableDescriptor(table.getBytes());

		return Lists.newArrayList(Iterables.transform(desc.getFamilies(), new Function<HColumnDescriptor, String>() {

			@Override

			public String apply(final HColumnDescriptor desc) {

				return desc.getNameAsString();

			}

		}));

	}

	public void truncateTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {

		final HBaseAdmin admin = getHBaseAdmin(clusterName);

		if (!admin.tableExists(table)) throw new IllegalStateException("cannot truncate unexisting table");

		final HTableDescriptor desc = admin.getTableDescriptor(table.getBytes());

		log.info("disabling table: " + table);

		admin.disableTable(table);

		log.info("deleting table: " + table);

		admin.deleteTable(table);

		log.info("creating table: " + table);

		admin.createTable(desc);

	}

	public boolean existTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {

		final HBaseAdmin admin = getHBaseAdmin(clusterName);

		return admin.tableExists(table);

	}

	public void dropTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {

		final HBaseAdmin admin = getHBaseAdmin(clusterName);

		if (!admin.tableExists(table)) throw new IllegalStateException("cannot drop unexisting table: '" + table + "'");

		log.info("disabling table: " + table);

		admin.disableTable(table);

		log.info("deleting table: " + table);

		admin.deleteTable(table);

	}

	public void createTable(final ClusterName clusterName, final String table, final Set<String> columns) throws IOException, HadoopServiceException {

		final HBaseAdmin admin = getHBaseAdmin(clusterName);

		if (admin.tableExists(table)) throw new IllegalStateException("table already exists");

		final HTableDescriptor desc = new HTableDescriptor(table);

		for (final String column : columns) {

			final HColumnDescriptor hds = new HColumnDescriptor(column);

			hds.setMaxVersions(getMaxVersions());

			desc.addFamily(hds);

		}

		log.info("creating hbase table: " + table);

		admin.createTable(desc);

		log.info("created hbase table: [" + table + "]");

		log.debug("descriptor: [" + desc.toString() + "]");

	}

	public void ensureTable(final ClusterName clusterName, final String table, final Set<String> columns) throws IOException, HadoopServiceException {

		final HBaseAdmin admin = getHBaseAdmin(clusterName);

		if (!admin.tableExists(table)) {

			createTable(clusterName, table, columns);

		} else {

			final HTableDescriptor desc = admin.getTableDescriptor(Bytes.toBytes(table));

			final Set<String> foundColumns = Sets.newHashSet(Iterables.transform(Lists.newArrayList(desc.getColumnFamilies()), HadoopUtils.columnName()));

			log.info("ensuring columns on table " + table + ": " + columns);

			final List<String> missingColumns = Lists.newArrayList(Iterables.filter(columns, HadoopUtils.columnPredicate(foundColumns)));

			if (!missingColumns.isEmpty()) {

				if (admin.isTableEnabled(table)) {

					admin.disableTable(table);

				}

				for (final String column : missingColumns) {

					log.info("hbase table: '" + table + "', adding column: " + column);

					admin.addColumn(table, new HColumnDescriptor(column));

				}

				admin.enableTable(table);

			}

		}

	}

	public void writeOnHBase(final ClusterName clusterName, final String tableName, final List<Put> puts) throws IOException {

		final Configuration conf = configurationEnumerator.get(clusterName);

		final HTable table = new HTable(conf, tableName);

		try {

			table.put(puts);

		} finally {

			table.flushCommits();

			table.close();

		}

	}

	public void deleteFromHBase(final ClusterName clusterName, final String tableName, final List<Delete> deletes) throws IOException {

		final Configuration conf = configurationEnumerator.get(clusterName);

		final HTable table = new HTable(conf, tableName);

		try {

			table.delete(deletes);

		} finally {

			table.flushCommits();

			table.close();

		}

	}

	public void deleteColumnsFromHBase(final ClusterName clusterName, final String tableName, final List<HBaseRowDescriptor> columns) throws IOException {

		final Configuration conf = configurationEnumerator.get(clusterName);

		final HTable table = new HTable(conf, tableName);

		try {

			final List<Delete> deletes = Lists.newArrayList();

			for(HBaseRowDescriptor desc : columns) {

				final Delete d = new Delete(Bytes.toBytes(desc.getRowKey()));

				for(Column c : desc.getColumns()) {

					for(String qualifier : c.getQualifier()) {

						log.info(String.format("delete from row '%s' cf '%s::%s'", desc.getRowKey(), c.getFamily(), qualifier));

						d.deleteColumn(Bytes.toBytes(c.getFamily()), Bytes.toBytes(qualifier));

					}

				}

				deletes.add(d);

			}

			table.delete(deletes);

		} finally {

			table.flushCommits();

			table.close();

		}

	}

	public Result getRow(final ClusterName clusterName, final String tableName, final byte[] id) throws IOException {

		final Configuration conf = configurationEnumerator.get(clusterName);

		final HTable table = new HTable(conf, tableName);

		try {

			return table.get(new Get(id));

		} finally {

			table.close();

		}

	}

	public Map<String, HBaseRowDescriptor> describeRows(final ClusterName clusterName, final String tableName, final List<String> rowKeys) throws IOException {

		final Map<String, HBaseRowDescriptor> map = Maps.newHashMap();

		for(String rowKey : rowKeys) {

			map.put(rowKey, describeRow(clusterName, tableName, rowKey));

		}

		return map;

	}

	public HBaseRowDescriptor describeRow(final ClusterName clusterName, final String tableName, final String rowKey) throws IOException {

		final Configuration conf = configurationEnumerator.get(clusterName);

		final HTable table = new HTable(conf, tableName);

		final HBaseRowDescriptor desc = new HBaseRowDescriptor();

		try {

			final Result r = table.get(new Get(Bytes.toBytes(rowKey)));

			if (r.isEmpty()) {

				return desc;

			}

			final List<Column> columns = Lists.newArrayList();

			for(Entry<byte[], NavigableMap<byte[], byte[]>> e : r.getNoVersionMap().entrySet()) {

				final Set<byte[]> qualifiers = e.getValue().keySet();

				final String family = new String(e.getKey());

				final Column col = new Column(family);

				for(byte[] q : qualifiers) {

					String qs = new String(q);

					col.getQualifier().add(qs);

				}

				columns.add(col);

			}

			desc.setColumns(columns);

			desc.setRowKey(rowKey);

			return desc;

		} finally {

			table.close();

		}

	}

	public List<Result> getRows(final ClusterName clusterName, final String tableName, final Scan scan) throws IOException {

		final Configuration conf = configurationEnumerator.get(clusterName);

		final HTable table = new HTable(conf, tableName);

		try {

			final ResultScanner rs = table.getScanner(scan);

			try {

				return Lists.newArrayList(rs.iterator());

			} finally {

				rs.close();

			}

		} finally {

			table.close();

		}

	}

	public boolean deleteFromHdfs(final ClusterName clusterName, final String path) throws HadoopServiceException {

		if (StringUtils.isBlank(path))

			throw new HadoopServiceException("Cannot deleteFromHBase an empty HDFS path.");

		final Configuration conf = configurationEnumerator.get(clusterName);

		try {

			final FileSystem hdfs = FileSystem.get(conf);

			final Path absolutePath = new Path(URI.create(conf.get("fs.defaultFS") + path));

			if (hdfs.exists(absolutePath)) {

				log.debug("deleteFromHBase path: " + absolutePath.toString());

				hdfs.delete(absolutePath, true);

				log.info("deleted path: " + absolutePath.toString());

				return true;

			} else {

				log.warn("cannot deleteFromHBase unexisting path: " + absolutePath.toString());

				return false;

			}

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	public boolean createHdfsDir(final ClusterName clusterName, final String path, final boolean force) throws HadoopServiceException {

		if (StringUtils.isBlank(path))

			throw new HadoopServiceException("Cannot create an empty HDFS path.");

		final Configuration conf = configurationEnumerator.get(clusterName);

		try {

			final FileSystem hdfs = FileSystem.get(conf);

			final Path absolutePath = new Path(URI.create(conf.get("fs.defaultFS") + path));

			if (!hdfs.exists(absolutePath)) {

				hdfs.mkdirs(absolutePath);

				log.info("created path: " + absolutePath.toString());

				return true;

			} else if (force) {

				log.info(String.format("found directory '%s', force delete it", absolutePath.toString()));

				hdfs.delete(absolutePath, true);

				hdfs.mkdirs(absolutePath);

				log.info("created path: " + absolutePath.toString());

				return true;

			} else {

				log.info(String.format("directory already exists: '%s', nothing to do", absolutePath.toString()));

				return false;

			}

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	public Configuration getClusterConiguration(final ClusterName clusterName) {

		return configurationEnumerator.get(clusterName);

	}

	public int getMaxVersions() {

		return maxVersions;

	}

	@Required

	public void setMaxVersions(final int maxVersions) {

		this.maxVersions = maxVersions;

	}

	public HadoopClientMap getClients() {

		return clients;

	}

}

package eu.dnetlib.data.hadoop.hbase;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.context.annotation.Profile;

import org.springframework.core.io.ClassPathResource;

import org.springframework.core.io.Resource;

import eu.dnetlib.data.hadoop.HadoopClientMap;

import eu.dnetlib.data.hadoop.HadoopServiceCore;

import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;

import eu.dnetlib.data.hadoop.config.ConfigurationFactory;

import eu.dnetlib.data.hadoop.oozie.OozieClientFactory;

import eu.dnetlib.data.mapreduce.JobClientFactory;

@Configuration

@Profile(value = "test")

public class HBaseTestContextConfiguration {

	public static final String ENABLED_CLIENTS = "{"

			+ "\"DM\":{\"oozie\":\"false\",\"mapred\":\"false\",\"hbase\":\"true\"},"

			+ "\"IIS\":{\"oozie\":\"false\",\"mapred\":\"false\",\"hbase\":\"false\"}"

			+ "}";

	public static final int MAX_VERSIONS = 10;

	@Bean

	public HadoopServiceCore hadoopServiceCore() {

		final HadoopServiceCore core = new HadoopServiceCore();

		core.setMaxVersions(MAX_VERSIONS);

		System.out.println("using hbase max versions: " + MAX_VERSIONS);

		return core;

	}

	@Bean(initMethod = "init")

	public HadoopClientMap hadoopClientMap() throws InterruptedException {

		final HadoopClientMap clientMap = new HadoopClientMap();

		clientMap.setEnabledClients(ENABLED_CLIENTS);

		clientMap.setClientsInitTime(10);

		return clientMap;

	}

	@Bean

	public HBaseAdminFactory hBaseAdminFactory() {

		return new HBaseAdminFactory();

	}

	@Bean

	public OozieClientFactory oozieClientFactory() {

		return new OozieClientFactory();

	}

	@Bean

	public JobClientFactory jobClientFactory() {

		return new JobClientFactory();

	}

	@Bean

	public ConfigurationEnumerator configurationEnumerator() {

		return new ConfigurationEnumerator();

	}

	@Bean

	public ConfigurationFactory DM() {

		return get(new ClassPathResource("/eu/dnetlib/data/hadoop/config/hadoop-default.dm.cnr.properties"));

	}

	@Bean

	public ConfigurationFactory IIS() {

		return get(new ClassPathResource("/eu/dnetlib/data/hadoop/config/hadoop-default.iis.icm.properties"));

	}

	protected ConfigurationFactory get(final Resource props) {

		final ConfigurationFactory configurationFactory = new ConfigurationFactory();

		configurationFactory.setDefaults(props);

		return configurationFactory;

	}

}

package eu.dnetlib.data.hadoop.oozie;

import java.util.Date;

import java.util.HashMap;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.oozie.client.OozieClient;

import org.apache.oozie.client.OozieClientException;

import org.apache.oozie.client.WorkflowJob.Status;

import eu.dnetlib.data.hadoop.action.JobCompletion;

import eu.dnetlib.data.hadoop.action.JobMonitor;

import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;

public class OozieJobMonitor extends JobMonitor {

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

	private final OozieClient oozieClient;

	private final String jobId;

	public OozieJobMonitor(OozieClient oozieClient, String jobId, JobCompletion callback) {

		super(callback);

		this.oozieClient = oozieClient;

		this.jobId = jobId;

	}

	@Override

	public void run() {

		try {

			log.info("waiting for oozie job completion: " + getHadoopId());

			Status status = doGetStatus();

			while (status.equals(Status.RUNNING)) {

				Thread.sleep(monitorSleepTimeSecs * 1000);

				Status currentStatus = doGetStatus();

				if (!status.equals(currentStatus)) {

					status = currentStatus;

					lastActivity = new Date(System.currentTimeMillis());

				}

			}

			log.debug("job " + jobId + " finihsed with status: " + status);

			if (status == Status.SUCCEEDED) {

				// TODO set some content to return to the blackboard msg.

				getCallback().done(new HashMap<String, String>());

            } else {

				// TODO retrieve some failure information from the oozie client.

				String msg = "hadoop job: " + getHadoopId() + " failed with status: " + getStatus() + ", oozie log:\n "

						+ getOozieClient().getJobLog(getHadoopId()) + "\n\n";

				getCallback().failed(msg, new HadoopServiceException(msg));

            }

		} catch (Throwable e) {

			getCallback().failed(getHadoopId(), e);

		}

	}

	@Override

	public String getHadoopId() {

		return jobId;

	}

	public OozieClient getOozieClient() {

		return oozieClient;

	}

	@Override

	public String getStatus() {

		try {

			return doGetStatus().toString();

		} catch (OozieClientException e) {

			log.error("error accessing job status", e);

			return "UNKNOWN";

		}

	}

	private Status doGetStatus() throws OozieClientException {

		return getOozieClient().getJobInfo(getHadoopId()).getStatus();

	}

	@Override

	public Date getLastActivity() {

		return lastActivity;

	}

	@Override

	public Date getStartTime() throws HadoopServiceException {

		try {

			return getOozieClient().getJobInfo(getHadoopId()).getStartTime();

		} catch (OozieClientException e) {

			throw new HadoopServiceException("unable to read job start time", e);

		}

	}

	@Override

	public String getTrackerUrl() {

		return getOozieClient().getOozieUrl();

	}

	@Override

	public void kill() {

		try {

			getOozieClient().kill(getHadoopId());

		} catch (OozieClientException e) {

			log.error("unable to kill job: " + getHadoopId(), e);

		}

	}

}

package eu.dnetlib.data.hadoop;

import java.io.IOException;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.Set;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.rmi.HadoopJobDescriptor;

import eu.dnetlib.data.hadoop.rmi.HadoopService;

import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;

import eu.dnetlib.data.hadoop.rmi.hbase.HBaseRowDescriptor;

import eu.dnetlib.enabling.tools.AbstractBaseService;

import eu.dnetlib.enabling.tools.blackboard.NotificationHandler;

import org.apache.hadoop.conf.Configuration;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

/**

 * The Class HadoopServiceImpl.

 */

public class HadoopServiceImpl extends AbstractBaseService implements HadoopService {

	/**

	 * notification handler.

	 */

	private NotificationHandler notificationHandler;

	/** The hadoop service core. */

	@Autowired

	private HadoopServiceCore hadoopServiceCore;

	/** The job registry. */

	@Autowired

	private JobRegistry jobRegistry;

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#listAvailableJobs()

	 */

	@Override

	public List<String> listAvailableJobs() throws HadoopServiceException {

		List<String> res = Lists.newArrayList();

		return res;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#listJobs(java.lang.String)

	 */

	@Override

	public List<HadoopJobDescriptor> listJobs(final String clusterName) throws HadoopServiceException {

		return jobRegistry.listJobs(checkExists(clusterName));

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#killJob(java.lang.String)

	 */

	@Override

	public boolean killJob(final String jobId) throws HadoopServiceException {

		jobRegistry.unregisterJob(jobId);

		return true;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.enabling.tools.AbstractBaseService#notify(java.lang.String, java.lang.String, java.lang.String, java.lang.String)

	 */

	@Override

	public void notify(final String subscriptionId, final String topic, final String isId, final String message) {

		getNotificationHandler().notified(subscriptionId, topic, isId, message);

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#listHbaseTables(java.lang.String)

	 */

	@Override

	public List<String> listHbaseTables(final String clusterName) throws HadoopServiceException {

		try {

			return hadoopServiceCore.listTables(checkExists(clusterName));

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#ensureHbaseTable(java.lang.String, java.lang.String, java.util.Set)

	 */

	@Override

	public boolean ensureHbaseTable(final String clusterName, final String tableName, final Set<String> columns) throws HadoopServiceException {

		try {

			hadoopServiceCore.ensureTable(checkExists(clusterName), tableName, columns);

			return true;

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#createHbaseTable(java.lang.String, java.lang.String, java.util.Set)

	 */

	@Override

	public boolean createHbaseTable(final String clusterName, final String tableName, final Set<String> columns) throws HadoopServiceException {

		try {

			hadoopServiceCore.createTable(checkExists(clusterName), tableName, columns);

			return true;

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#truncateHbaseTable(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean truncateHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {

		try {

			hadoopServiceCore.truncateTable(checkExists(clusterName), tableName);

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

		return true;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#dropHbaseTable(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean dropHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {

		try {

			hadoopServiceCore.dropTable(checkExists(clusterName), tableName);

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

		return true;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#existHbaseTable(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean existHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {

		try {

			return hadoopServiceCore.existTable(checkExists(clusterName), tableName);

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.hadoop.rmi.HadoopService#getClusterConfiguration(java.lang.String)

	 */

	@Override

	public Map<String, String> getClusterConfiguration(final String clusterName) throws HadoopServiceException {

		final Configuration conf = hadoopServiceCore.getClusterConiguration(checkExists(clusterName));

		final Map<String, String> res = Maps.newHashMap();

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

			res.put(e.getKey(), e.getValue());

		}

		return res;

	}

	@Override

	public boolean deleteHdfsPath(final String clusterName, final String path) throws HadoopServiceException {

		return hadoopServiceCore.deleteFromHdfs(checkExists(clusterName), path);

	}

	@Override

	public boolean createHdfsDirectory(final String clusterName, final String path, final boolean force) throws HadoopServiceException {

		return hadoopServiceCore.createHdfsDir(checkExists(clusterName), path, force);

	}

	@Override

	public List<String> listClusters() throws HadoopServiceException {

		try {

			return ClusterName.asStringList();

		} catch (Throwable e) {

			throw new HadoopServiceException(e);

		}

	}

	@Override

	public List<String> describeHbaseTable(final String clusterName, final String tableName) throws HadoopServiceException {

		try {

			return hadoopServiceCore.describeTable(checkExists(clusterName), tableName);

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	@Override

	public HBaseRowDescriptor describeHBaseColumn(final String clusterName, final String tableName, final String rowKey) throws HadoopServiceException {

		try {

			return hadoopServiceCore.describeRow(checkExists(clusterName), tableName, rowKey);

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	@Override

	public Map<String, HBaseRowDescriptor> describeHBaseColumns(final String clusterName, final String tableName, final List<String> rowKeys) throws HadoopServiceException {

		try {

			return hadoopServiceCore.describeRows(checkExists(clusterName), tableName, rowKeys);

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	@Override

	public boolean deleteHBaseColumn(final String clusterName, final String tableName, final HBaseRowDescriptor column) throws HadoopServiceException {

		try {

			hadoopServiceCore.deleteColumnsFromHBase(checkExists(clusterName), tableName, Lists.newArrayList(column));

			return true;

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	@Override

	public boolean deleteHBaseColumns(final String clusterName, final String tableName, final List<HBaseRowDescriptor> column) throws HadoopServiceException {

		try {

			hadoopServiceCore.deleteColumnsFromHBase(checkExists(clusterName), tableName, column);

			return true;

		} catch (IOException e) {

			throw new HadoopServiceException(e);

		}

	}

	///////////////////

	/**

	 * Check exists.

	 *

	 * @param clusterName

	 *            the cluster name

	 * @return the cluster name

	 * @throws HadoopServiceException

	 *             the hadoop service exception

	 */

	private ClusterName checkExists(final String clusterName) throws HadoopServiceException {

		try {

			return ClusterName.valueOf(clusterName);

		} catch (final IllegalArgumentException e) {

			throw new HadoopServiceException("Invalid cluster name: " + clusterName);

		}

	}

	/**

	 * Gets the notification handler.

	 *

	 * @return the notification handler

	 */

	public NotificationHandler getNotificationHandler() {

		return notificationHandler;

	}

	/**

	 * Sets the notification handler.

	 *

	 * @param notificationHandler

	 *            the new notification handler

	 */

	@Required

	public void setNotificationHandler(final NotificationHandler notificationHandler) {

		this.notificationHandler = notificationHandler;

	}

}

package eu.dnetlib.data.hadoop.utils;

import java.io.ByteArrayOutputStream;

import java.io.DataOutputStream;

import java.io.IOException;

import java.util.Map;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.apache.hadoop.hbase.filter.PrefixFilter;

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

import org.dom4j.Document;

import org.dom4j.Node;

public class ScanFactory {

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

	public static String getScan(final ScanProperties scanProperties) throws IOException {

		Scan scan = new Scan();

		scan.setCaching(scanProperties.getCaching());

		scan.setCacheBlocks(false); // don't set to true for MR jobs

		scan.setFilter(scanProperties.getFilterList());

		for (String family : scanProperties.getFamilies()) {

			scan.addFamily(family.getBytes());

		}

		log.debug("serializing scan");

		return convertScanToString(scan);

	}

	public static ScanProperties parseScanProperties(final Document doc, final Map<String, String> bbParams) {

		log.debug("setting job scanner");

		ScanProperties scanProperties = new ScanProperties(doc.valueOf("//FILTERS/@operator"));

		String caching = doc.valueOf("//SCAN/@caching");

		if (!StringUtils.isBlank(caching)) {

			log.info("overriding default scan caching with: " + caching);

			scanProperties.setCaching(Integer.valueOf(caching));

		}

		for (Object o : doc.selectNodes("//SCAN/FAMILIES/FAMILY")) {

			Node node = (Node) o;

			String value = node.valueOf("./@value");

			if ((value == null) || value.isEmpty()) {

				value = bbParams.get(node.valueOf("./@param"));

			}

			log.debug("scanner family value: " + value);

			scanProperties.getFamilies().add(value);

		}

		for (Object o : doc.selectNodes("//SCAN/FILTERS/FILTER")) {

			Node node = (Node) o;

			String filterType = node.valueOf("./@type");

			String value = node.valueOf("./@value");

			if ((value == null) || value.isEmpty()) {

				value = bbParams.get(node.valueOf("./@param"));

			}

			if (filterType.equals("prefix")) {

				log.debug("scanner prefix filter, value: " + value);

				scanProperties.getFilterList().addFilter(new PrefixFilter(value.getBytes()));

			} // TODO add more filterType cases here

		}

		return scanProperties;

	}

	/**

	 * Writes the given scan into a Base64 encoded string.

	 *

	 * @param scan

	 *            The scan to write out.

	 * @return The scan saved in a Base64 encoded string.

	 * @throws IOException

	 *             When writing the scan fails.

	 */

	private static String convertScanToString(final Scan scan) throws IOException {

		ByteArrayOutputStream out = new ByteArrayOutputStream();

		DataOutputStream dos = new DataOutputStream(out);

		scan.write(dos);

		return Base64.encodeBytes(out.toByteArray());

	}

}

package eu.dnetlib.data.hadoop.hbase;

import java.io.IOException;

import java.util.Arrays;

import java.util.List;

import com.google.common.base.Predicates;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;

import eu.dnetlib.data.transform.Row;

import eu.dnetlib.data.transform.XsltRowTransformerFactory;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;

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

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

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

/**

 * The Class HbaseTableFeeder provides abstraction to ship batch operation on an HBase table

 */

public abstract class HbaseTableFeeder {

	/**

	 * The Constant log.

	 */

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

	/**

	 * The configuration enumerator.

	 */

	@Autowired

	protected ConfigurationEnumerator configurationEnumerator;

	/**

	 * The batch size.

	 */

	private int batchSize = 100;

	/**

	 * The result set client factory.

	 */

	private ResultSetClientFactory resultSetClientFactory;

	/**

	 * Adds the operation.

	 *

	 * @param buffer the buffer

	 * @param row    the row

	 */

	protected abstract void addOperation(final List<Mutation> buffer, final Row row);

	/**

	 * Feed.

	 *

	 * @param epr         the epr

	 * @param xsl         the xsl

	 * @param clusterName the cluster name

	 * @param tableName   the table name

	 * @param simulation  the simulation

	 * @return the int

	 * @throws IOException          Signals that an I/O exception has occurred.

	 * @throws InterruptedException the interrupted exception

	 */

	public int feed(final String epr, final String xsl, final ClusterName clusterName, final String tableName, final boolean simulation)

			throws IOException, InterruptedException {

		return doWrite(asRows(epr, xsl), getConf(clusterName), tableName, simulation);

	}

	/**

	 * Do writeOnHBase.

	 *

	 * @param rows          the rows

	 * @param configuration the configuration

	 * @param tableName     the table name

	 * @param simulation    the simulation

	 * @return the int

	 * @throws IOException          Signals that an I/O exception has occurred.

	 * @throws InterruptedException the interrupted exception

	 */

	private int doWrite(final Iterable<Row> rows, final Configuration configuration, final String tableName, final boolean simulation)

			throws IOException, InterruptedException {

		final List<Mutation> buffer = Lists.newArrayList();

		int count = 0;

		if (simulation) {

			log.info("running in simulation mode ...");

			log.info(String.format("... simulated import of %d records", Iterables.size(rows)));

		} else {

			final HTable htable = new HTable(configuration, tableName);

			try {

				int i = 0;

				for (final Row row : rows) {

					addOperation(buffer, row);

					if ((++i % getBatchSize()) == 0) {

						flush(tableName, buffer, htable);

						count += buffer.size();

						buffer.clear();

					}

				}

			} finally {

				if (!buffer.isEmpty()) {

					flush(tableName, buffer, htable);

					count += buffer.size();

				}

				htable.flushCommits();

				htable.close();

			}

		}

		return count;

	}

	private void flush(final String tableName, final List<Mutation> buffer, final HTable htable) throws IOException, InterruptedException {

		if (!checkOp(htable.batch(buffer), tableName)) throw new IOException("unable to flush operation on HBase table: " + tableName);

	}

	private boolean checkOp(final Object[] res, final String tableName) throws IOException {

		return Iterables.all(Arrays.asList(res), Predicates.notNull());

	}

	/**

	 * As rows.

	 *

	 * @param epr the epr

	 * @param xsl the xsl

	 * @return the iterable

	 */

	protected Iterable<Row> asRows(final String epr, final String xsl) {

		return Iterables.concat(Iterables.transform(getResultSetClientFactory().getClient(epr), XsltRowTransformerFactory.newInstance(xsl)));

	}

	/**

	 * Gets the conf.

	 *

	 * @param clusterName the cluster name

	 * @return the conf

	 */

	protected Configuration getConf(final ClusterName clusterName) {

		return configurationEnumerator.get(clusterName);

	}

	/**

	 * Gets the batch size.

	 *

	 * @return the batch size

	 */

	public int getBatchSize() {

		return batchSize;

	}

	/**

	 * Sets the batch size.

	 *

	 * @param batchSize the new batch size

	 */

	public void setBatchSize(final int batchSize) {

		this.batchSize = batchSize;

	}

	/**

	 * Gets the result set client factory.

	 *

	 * @return the result set client factory

	 */

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	/**

	 * Sets the result set client factory.

	 *

	 * @param resultSetClientFactory the new result set client factory

	 */

	@Required

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

}

package eu.dnetlib.data.hadoop.utils;

import java.util.Map;

import java.util.Set;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import eu.dnetlib.data.hadoop.rmi.HadoopJobType;

public class JobProfile {

	private final Map<String, String> jobDefinition = Maps.newHashMap();

	private final Set<String> requiredParams = Sets.newHashSet();

	private ScanProperties scanProperties;

	private String name;

	private String description = "";

	private HadoopJobType jobType;

	public boolean isEmpty() {

		return getJobDefinition().isEmpty();

	}

	public Map<String, String> getJobDefinition() {

		return jobDefinition;

	}

	public Set<String> getRequiredParams() {

		return requiredParams;

	}

	public String getName() {

		return name;

	}

	public void setName(String name) {

		this.name = name;

	}

	public String getDescription() {

		return description;

	}

	public void setDescription(String description) {

		this.description = description;

	}

	public ScanProperties getScanProperties() {

		return scanProperties;

	}

	public void setScanProperties(ScanProperties scanProperties) {

		this.scanProperties = scanProperties;

	}

	public HadoopJobType getJobType() {

		return jobType;

	}

	public void setJobType(HadoopJobType jobType) {

		this.jobType = jobType;

	}

}

package eu.dnetlib.data.hadoop.hbase;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

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

import eu.dnetlib.data.transform.Column;

import eu.dnetlib.data.transform.Row;

/**

 * The Class HBaseDeleteFeeder performs a batch of Delete operations.

 */

public class HBaseDeleteFeeder extends HbaseTableFeeder {

	/**

	 * Logger.

	 */

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

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.hadoop.hbase.HbaseTableFeeder#addOperation(java.util.List, eu.dnetlib.data.transform.Row)

	 */

	@Override

	protected void addOperation(final List<Mutation> buffer, final Row row) {

		final Delete delete = new Delete(Bytes.toBytes(row.getKey()));

		delete.setWriteToWAL(true);

		for (final Column<String, byte[]> col : row) {

			log.debug(String.format("deleting K: '%s' CF:'%s' Q:'%s'", row.getKey(), row.getColumnFamily(), col.getName()));

			delete.deleteColumns(Bytes.toBytes(row.getColumnFamily()), Bytes.toBytes(col.getName()));

		}

		buffer.add(delete);

	}

}

package eu.dnetlib.data.hadoop;

import java.util.Date;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Required;

import com.google.common.collect.BiMap;

import com.google.common.collect.HashBiMap;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import eu.dnetlib.data.hadoop.HadoopJob.Status;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.rmi.HadoopJobDescriptor;

import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;

import eu.dnetlib.data.hadoop.utils.HadoopUtils;

public class JobRegistry {

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

	private int maxJobs;

	private final BiMap<String, HadoopJob> jobs = HashBiMap.create();

	public String registerJob(HadoopJob hadoopJob) throws HadoopServiceException {

		if (jobs.containsValue(hadoopJob)) { return jobs.inverse().get(hadoopJob); }

		if (jobs.size() > getMaxJobs()) {

			removeOldestProcess();

		}

		jobs.put(hadoopJob.getId(), hadoopJob);

		log.info("Registered hadoop job " + hadoopJob.getId());

		hadoopJob.startMonitor();

		return hadoopJob.getId();

	}

	public Status getJobStatus(String id) {

		return findJob(id).getStatus();

	}

	public HadoopJob findJob(String id) {

		return jobs.get(id);

	}

	public void unregisterJob(String id) throws HadoopServiceException {

		if (!jobs.containsKey(id)) { throw new HadoopServiceException("unable to unregister job, could not find jobId in registry: " + id); }

		log.info("unregistering job: " + id);

		jobs.get(id).getJobMonitor().kill();

		jobs.remove(id);

	}

	private void removeOldestProcess() throws HadoopServiceException {

		Date oldDate = new Date();

		String oldId = null;

		for (Entry<String, HadoopJob> e : jobs.entrySet()) {

			final HadoopJob hadoopJob = e.getValue();

			if (hadoopJob.isComplete()) {

				final Date date = hadoopJob.getLastActivity();

				if (date.before(oldDate)) {

					oldDate = date;

					oldId = e.getKey();

				}

			}

		}

		if (oldId != null) {

			unregisterJob(oldId);

		}

	}

	public List<HadoopJobDescriptor> listJobs(ClusterName clusterName) {

		Map<String, HadoopJob> filtered = Maps.filterValues(jobs, HadoopUtils.filterByCluster(clusterName));

		return Lists.newArrayList(Iterables.transform(filtered.entrySet(), HadoopUtils.hadoopJobDescriptor()));

	}

	@Required

	public void setMaxJobs(final int maxJobs) {

		this.maxJobs = maxJobs;

	}

	public int getMaxJobs() {

		return maxJobs;

	}

}

package eu.dnetlib.data.hadoop.hbase;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;