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>dnet45-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.7.6</version>

	<scm>

	    <developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-hadoop-service/tags/dnet-hadoop-service-2.7.6</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-hadoop-commons</artifactId>

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

		</dependency>

<!--

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>

-->

		<dependency>

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

			<artifactId>hbase</artifactId>

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

			<exclusions>

                                <exclusion>

                                        <artifactId>jasper-compiler</artifactId>

                                        <groupId>tomcat</groupId>

                                </exclusion>

                                <exclusion>

                                        <artifactId>jasper-runtime</artifactId>

                                        <groupId>tomcat</groupId>

                                </exclusion>

                                <exclusion>

                                        <artifactId>jamon-runtime</artifactId>

                                        <groupId>org.jamon</groupId>

                                </exclusion>

				<exclusion>

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

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

				</exclusion>

                                <exclusion>

                                        <artifactId>jsp-2.1</artifactId>

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

                                </exclusion>

				<exclusion>

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

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

				</exclusion>

                                <exclusion>

                                        <artifactId>jetty</artifactId>

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

                                </exclusion>

                                <exclusion>

                                        <artifactId>jetty-util</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>

				<exclusion>

					<artifactId>asm</artifactId>

					<groupId>asm</groupId>

				</exclusion>

				<exclusion>

					<artifactId>jaxb-api</artifactId>

					<groupId>javax.xml.bind</groupId>

				</exclusion>

				<exclusion>

					<artifactId>jaxb-impl</artifactId>

					<groupId>com.sun.xml.bind</groupId>

				</exclusion>

				<exclusion>

					<artifactId>commons-codec</artifactId>

					<groupId>commons-codec</groupId>

				</exclusion>

                                <exclusion>

                                        <artifactId>commons-logging</artifactId>

                                        <groupId>commons-logging</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>

                <dependency>

                        <groupId>commons-codec</groupId>

                        <artifactId>commons-codec</artifactId>

                        <version>${commons.codec.version}</version>

                </dependency>

                <dependency>

                        <groupId>commons-net</groupId>

                        <artifactId>commons-net</artifactId>

                        <version>${commons.net.version}</version>

                </dependency>

		<dependency>

                  <groupId>org.apache.commons</groupId>

                  <artifactId>commons-compress</artifactId>

                  <version>${commons.compress.version}</version>

		</dependency>

	</dependencies>

</project>

services.hadoop.clients={"DM":{"oozie":"true","mapred":"true","hbase":"true"},"IIS":{"oozie":"true","mapred":"false","hbase":"false"}}

services.hadoop.hbase.tablefeeder.batchsize=500

services.hadoop.jobregistry.size=100

services.hadoop.lib.path=/user/dnet/lib/dnet-mapreduce-jobs-assembly-0.0.6.2-SNAPSHOT.jar

services.hadoop.hbase.maxversions=1

services.hadoop.clients.init.timeout=30

services.hadoop.user=dnet

package eu.dnetlib.data.hadoop.oozie;

import java.io.IOException;

import java.util.*;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringUtils;

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.WorkflowAction;

import org.apache.oozie.client.WorkflowJob;

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;

import static java.lang.String.format;

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 static final String ACTION_TYPE_SUBWORKFLOW = "sub-workflow";

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

	@Deprecated

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

		super(callback);

		this.oozieClient = oozieClient;

		this.jobId = jobId;

	}

	public OozieJobMonitor(final OozieClient oozieClient, String jobId, final JobCompletion callback, final Set<String> workflowActions) {

		super(callback);

		this.oozieClient = oozieClient;

		this.jobId = jobId;

		this.workflowActions = workflowActions;

	}

	@Override

	public void run() {

		try {

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

			Status status = Status.PREP;

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

				Thread.sleep(monitorSleepTimeSecs * 1000);

				try {

					final Status currentStatus = doGetStatus();

					if (!currentStatus.equals(status)) {

						status = currentStatus;

						lastActivity = new Date();

					}

				} catch (Throwable e) {

					log.warn(format("error polling status for job %s", jobId), e);

				}

			}

			log.debug(format("job %s finihsed with status: %s", jobId, status));

			if (Status.SUCCEEDED.equals(status)) {

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

				log.info(format("looking for oozie job(%s) output values: %s", getHadoopId(), workflowActions));

				final Properties report = getReport(getOozieClient(), getHadoopId(), workflowActions);

				if (report != null) {

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

					report.forEach((k, v) -> map.put(k.toString(), v.toString()));

					log.info("found oozie job report, size: " + map.size());

					getCallback().done(map);

				} else {

					log.warn("cannot find oozie job report!");

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

				}

            } else {

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

				String msg = format("hadoop job: %s failed with status: %s, oozie log:\n %s\n", getHadoopId(), getStatus(), getOozieClient().getJobLog(getHadoopId()));

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

            }

		} catch (Throwable e) {

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

		}

	}

	/**

	 * Provides report entries when found for given oozie job identifier. Returns null when report not found.

	 */

	private static Properties getReport(final OozieClient oozieClient, final String oozieJobId, final Set<String> workflowActions) throws OozieClientException, IOException {

		WorkflowJob oozieJob = oozieClient.getJobInfo(oozieJobId);

		for (WorkflowAction currentAction : oozieJob.getActions()) {

			log.info(String.format("looking for workflow actions to report, current: '%s'", currentAction.getName()));

			if (workflowActions.contains(currentAction.getName())) {

				log.info(String.format("found workflow action %s", currentAction.getName()));

				if (ACTION_TYPE_SUBWORKFLOW.equals(currentAction.getType())) {

					log.info(String.format("looking for sub-workflow actions external id: %s", currentAction.getExternalId()));

					Properties subworkflowProperties = getReport(oozieClient, currentAction.getExternalId(), workflowActions);

					if (subworkflowProperties != null) {

						return subworkflowProperties;

					}

				} else if (StringUtils.isNotBlank(currentAction.getData())) {

					Properties properties = new Properties();

					properties.load(IOUtils.toInputStream(currentAction.getData()));

					log.info(String.format("found workflow action(%s) properties size %s", currentAction.getName(), properties.values().size()));

					return properties;

				}

			} else {

				log.info(String.format("cannot find workflow action(%s) properties", currentAction.getName()));

			}

		}

		return null;

	}

	@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.oozie;

import eu.dnetlib.data.hadoop.AbstractHadoopClient;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.oozie.client.OozieClient;

/**

 * Factory bean for Oozie client instances.

 * 

 * oozie server may not be available in every hadoop deployment, so every oozie specific operation should be defined as optional.

 * 

 * @author claudio

 * 

 */

public class OozieClientFactory extends AbstractHadoopClient {

	private static final String ENV_ATTRIBUTE_OOZIE_SERVICE_LOC = "oozie.service.loc";

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

	public OozieClient newInstance(ClusterName clusterName) throws HadoopServiceException {

		final String oozieServiceLocation = configurationEnumerator.get(clusterName).get(ENV_ATTRIBUTE_OOZIE_SERVICE_LOC);

		log.info("init oozie client, cluster: " + clusterName.toString() + ", oozie server: " + oozieServiceLocation);

		setHadoopUser();

		try {

			return new OozieClient(oozieServiceLocation);

		} catch (Throwable e) {

			throw new HadoopServiceException("unable to initialize oozie client for cluster: " + clusterName.toString(), e);

		}

	}

}

package eu.dnetlib.data.hadoop.utils;

import java.util.Map;

import java.util.Map.Entry;

import org.apache.commons.math.stat.descriptive.SummaryStatistics;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;

import org.junit.Before;

import org.junit.Ignore;

import org.junit.Test;

import org.springframework.core.io.ClassPathResource;

import com.google.common.collect.Maps;

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

import eu.dnetlib.data.hadoop.hdfs.SequenceFileUtils;

import eu.dnetlib.miscutils.collections.Pair;

public class ReadSequenceFileTest {

	private static final Path SEQUENCE_FILE_PATH = new Path("hdfs://nmis-hadoop-cluster/tmp/indexrecords_db_openaireplus_sesam_SESAMALL.seq");

	private final String HADOOP_CONF_FILE = "/eu/dnetlib/data/hadoop/config/hadoop-default.dm.cnr.properties";

	private Configuration conf;

	@Before

	public void setUp() {

		final ConfigurationFactory confFactory = new ConfigurationFactory();

		confFactory.setDefaults(new ClassPathResource(HADOOP_CONF_FILE));

		conf = confFactory.getConfiguration();

	}

	@Test

	@Ignore

	public void testReadSequenceFile() throws Exception {

		final SummaryStatistics statsAll = new SummaryStatistics();

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

		int i = 0;

		for (Pair<Text, Text> pair : SequenceFileUtils.read(SEQUENCE_FILE_PATH, conf)) {

			final String id = pair.getKey().toString();

			final String record = pair.getValue().toString();

			final int length = record.getBytes().length;

			final String type = id.substring(0, 2);

			if (!stats.containsKey(type)) {

				stats.put(type, new SummaryStatistics());

			}

			statsAll.addValue(length);

			stats.get(type).addValue(length);

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

				System.out.println("Read " + i);

			}

		}

		printStats("ALL", statsAll);

		for (Entry<String, SummaryStatistics> e : stats.entrySet()) {

			printStats(e.getKey(), e.getValue());

		}

	}

	private void printStats(final String type, final SummaryStatistics stats) {

		System.out.println("************************************");

		System.out.println("Type: " + type);

		System.out.println(String.format("\tmin    : %.2f KBytes", stats.getMin() / 1024));

		System.out.println(String.format("\tmax    : %.2f KBytes", stats.getMax() / 1024));

		System.out.println(String.format("\tavg    : %.2f KBytes", stats.getMean() / 1024));

		System.out.println(String.format("\tstdDev : %.2f", stats.getStandardDeviation() / 1024));

	}

}

{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-hadoop-service/trunk/", "deploy_repository": "dnet45-snapshots", "version": "4", "mail": "sandro.labruzzo@isti.cnr.it,michele.artini@isti.cnr.it, claudio.atzori@isti.cnr.it, alessia.bardi@isti.cnr.it", "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet45-snapshots", "name": "dnet-hadoop-service"}

package eu.dnetlib.data.hadoop;

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

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

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public abstract class AbstractHadoopClient {

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

	@Autowired

	private ISClient isClient;

	@Autowired

	protected ConfigurationEnumerator configurationEnumerator;

	protected void setHadoopUser() throws HadoopServiceException {

		setHadoopUser(getDefaultUser());

	}

	private String getDefaultUser() throws HadoopServiceException {

		try {

			return isClient.queryForServiceProperty("default.hadoop.user");

		} catch (ISLookUpException e) {

			throw new HadoopServiceException(e);

		}

	}

	protected void setHadoopUser(final String userName) {

		log.info("setting HADOOP_USER_NAME as " + userName);

		System.setProperty("HADOOP_USER_NAME", userName);

	}

}

package eu.dnetlib.data.hadoop;

import java.io.IOException;

import java.util.Map;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

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

import eu.dnetlib.data.hadoop.hbase.HBaseAdminFactory;

import eu.dnetlib.data.hadoop.mapred.JobClientFactory;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.apache.hadoop.mapred.JobClient;

import org.apache.oozie.client.OozieClient;

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

public class HadoopClientMap {

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

	private JobClientFactory jobClientFactory;

	private OozieClientFactory oozieClientFactory;

	private HBaseAdminFactory hbaseAdminFactory;

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

	public boolean isMapreduceAvailable(final ClusterName name) {

		return isClientAvailable(name, "mapred");

	}

	public boolean isOozieAvailable(final ClusterName name) {

		return isClientAvailable(name, "oozie");

	}

	public boolean isHbaseAvailable(final ClusterName name) {

		return isClientAvailable(name, "hbase");

	}

	private boolean isClientAvailable(final ClusterName name, final String clientName) {

		final String clusterName = name.toString();

		return enabledClients.containsKey(clusterName) && "true".equals(enabledClients.get(clusterName).get(clientName));

	}

	public JobClient getJtClient(final ClusterName clusterName, final String username) throws HadoopServiceException, IOException {

		if (!isMapreduceAvailable(clusterName)) {

			throw new HadoopServiceException("jobtracker is not available for cluster " + clusterName.toString());

		}

		return getJobClientFactory().newInstance(clusterName, username);

	}

	public JobClient getJtClient(final ClusterName clusterName) throws HadoopServiceException, IOException {

		if (!isMapreduceAvailable(clusterName)) {

			throw new HadoopServiceException("jobtracker is not available for cluster " + clusterName.toString());

		}

		return getJobClientFactory().newInstance(clusterName);

	}

	public OozieClient getOozieClient(final ClusterName name) throws HadoopServiceException {

		if (!isOozieAvailable(name)) {

			throw new HadoopServiceException("oozie is not available for cluster " + name.toString());

		}

		return getOozieClientFactory().newInstance(name);

	}

	public HBaseAdmin getHbaseAdmin(final ClusterName name) throws HadoopServiceException {

		if (!isHbaseAvailable(name)) {

			throw new HadoopServiceException("hbase is not available for cluster " + name.toString());

		}

		return getHbaseAdminFactory().newInstance(name);

	}

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

	public String getEnabledClients() {

		return new Gson().toJson(enabledClients);

	}

	@Required

	@SuppressWarnings("unchecked")

	public void setEnabledClients(final String enabledClients) {

		this.enabledClients = new Gson().fromJson(enabledClients, Map.class);

	}

	public JobClientFactory getJobClientFactory() {

		return jobClientFactory;

	}

	@Required

	public void setJobClientFactory(final JobClientFactory jobClientFactory) {

		this.jobClientFactory = jobClientFactory;

	}

	public OozieClientFactory getOozieClientFactory() {

		return oozieClientFactory;

	}

	@Required

	public void setOozieClientFactory(final OozieClientFactory oozieClientFactory) {

		this.oozieClientFactory = oozieClientFactory;

	}

	public HBaseAdminFactory getHbaseAdminFactory() {

		return hbaseAdminFactory;

	}

	@Required

	public void setHbaseAdminFactory(final HBaseAdminFactory hbaseAdminFactory) {

		this.hbaseAdminFactory = hbaseAdminFactory;

	}

}

package eu.dnetlib.data.hadoop;

import java.util.List;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Iterables;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

public class ISClient {

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	public String getJobProfile(final String jobName) throws ISLookUpException {

		return serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(

				"/RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='" + jobName + "']");

	}

	public String queryForServiceProperty(final String key) throws ISLookUpException {

		return getServiceConfigValue(

				String.format(

						"distinct-values("

								+ "for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='HadoopServiceResourceType'] "

								+ "return $x//SERVICE_PROPERTIES/PROPERTY[./@ key='%s']/@value/string())",

						key));

	}

	public void updateCountElement(final String jobName, final String element, final String delta) {

		final String xquery =

				"let $x := //RESOURCE_PROFILE[" + ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='" + jobName

						+ "'], $tot := $x//STATUS/" + element + "/@value/number() " + delta + " return update replace $x//STATUS/" + element + " with <"

						+ element + " value='{$tot}' />";

		executeXUpdate(xquery);

	}

	public void updateDate(final String jobName) {

		log.info("increment last submission date for job: " + jobName);

		executeXUpdate("for $x in collection('')/RESOURCE_PROFILE["

				+ ".//RESOURCE_TYPE/@value='HadoopJobConfigurationDSResourceType' and .//HADOOP_JOB/@name='"

				+ jobName + "'] " + " return update value $x//LAST_SUBMISSION_DATE/@value with '" + DateUtils.now_ISO8601() + "' ");

	}

	private String getServiceConfigValue(final String xquery) throws ISLookUpException {

		log.debug("quering for service property: " + xquery);

		final List<String> urls = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

		if ((urls == null) || (urls.size() != 1))

			throw new IllegalStateException("unable to find unique service property, xquery: " + xquery);

		return Iterables.getOnlyElement(urls);

	}

	private boolean executeXUpdate(final String xupdate) {

		try {

			log.debug("running xupdate: " + xupdate);

			return serviceLocator.getService(ISRegistryService.class).executeXUpdate(xupdate);

		} catch (final ISRegistryException e) {

			log.error("unable to run xupdate: " + xupdate, e);

			return false;

		}

	}

}

package eu.dnetlib.data.hadoop.hbase;

import eu.dnetlib.data.hadoop.AbstractHadoopClient;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public class HBaseAdminFactory extends AbstractHadoopClient {

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

	public HBaseAdmin newInstance(final ClusterName clusterName) throws HadoopServiceException {

		try {

			log.info("init hbaseAdmin, cluster: " + clusterName.toString());

			setHadoopUser();

			return new HBaseAdmin(configurationEnumerator.get(clusterName));

		} catch (final Throwable e) {

			throw new HadoopServiceException("unable to initialize hbase client for cluster: " + clusterName.toString(), e);

		}

	}

}

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.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.hbase;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

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

import eu.dnetlib.data.transform.Column;

import eu.dnetlib.data.transform.Row;

/**

 * The Class HBasePutFeeder performs a batch of Put operations..

 */

public class HBasePutFeeder extends HbaseTableFeeder {

	/**

	 * Logger.

	 */

	private static final Log log = LogFactory.getLog(HBasePutFeeder.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 Put put = new Put(Bytes.toBytes(row.getKey()));

		put.setWriteToWAL(true);

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

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

			put.add(Bytes.toBytes(row.getColumnFamily()), Bytes.toBytes(col.getName()), col.getValue());

		}

		buffer.add(put);

	}

}

package eu.dnetlib.data.hadoop;

import java.io.IOException;

import java.net.URI;

import java.util.*;

import java.util.Map.Entry;

import java.util.stream.Collectors;

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.rmi.hbase.schema.HBaseTableDescriptor;

import eu.dnetlib.data.hadoop.rmi.hbase.schema.HBaseTableRegionInfo;

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.HRegionInfo;

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 {

		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {

			return Arrays.asList(admin.listTables())

					.stream()

					.map(HTableDescriptor::getNameAsString)

					.collect(Collectors.toList());

		}

	}

	public String getHBaseTableDescriptor(final ClusterName clusterName, final String tableName) throws HadoopServiceException, IOException {

		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {

			if (StringUtils.isBlank(tableName)) throw new HadoopServiceException("Table name cannot be empty or null");

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

			final List<HRegionInfo> tableRegions = admin.getTableRegions(tableName.getBytes());

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

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

			for (HColumnDescriptor hColDesc : Arrays.asList(desc.getColumnFamilies())) {

				columns.add(hColDesc.getNameAsString());

			}

			HBaseTableDescriptor htDescriptor = new HBaseTableDescriptor();

			htDescriptor.setColumns(columns);

			List<HBaseTableRegionInfo> regions = Lists.newArrayList();

			for (HRegionInfo info : tableRegions) {

				regions.add(new HBaseTableRegionInfo(new String(info.getStartKey()), new String(info.getEndKey())));

			}

			htDescriptor.setRegions(regions);

			if (log.isDebugEnabled()) {

				log.info("got configuration for table '" + tableName + "': " + htDescriptor.toString());

			}

			return htDescriptor.toString();

		}

	}

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

		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {

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

			return desc.getFamilies().stream()

					.map(d -> d.getNameAsString())

					.collect(Collectors.toList());

		}

	}

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

		try(final HBaseAdmin admin = clients.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 {

		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {

			return admin.tableExists(table);

		}

	}

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

		try(final HBaseAdmin admin = clients.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 String tableConfiguration) throws IOException, HadoopServiceException {

		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {

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

			if (StringUtils.isBlank(tableConfiguration)) throw new HadoopServiceException("empty table configuration");

			final HBaseTableDescriptor tableConf = HBaseTableDescriptor.fromJSON(tableConfiguration);

			doCreateTable(clusterName, table, tableConf.getColumns(), tableConf.getRegions());

		}

	}

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

		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {

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

			doCreateTable(clusterName, table, columns, null);

		}

	}

	public void doCreateTable(final ClusterName clusterName, final String table, final Set<String> columns, final List<HBaseTableRegionInfo> regions)

			throws IOException, HadoopServiceException {

		try(final HBaseAdmin admin = clients.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);

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

				log.debug(String.format("create using %s regions: %s", regions.size(), regions));

				admin.createTable(desc, getSplitKeys(regions));

			} else {

				admin.createTable(desc);

			}

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

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

		}

	}

	private byte[][] getSplitKeys(final List<HBaseTableRegionInfo> regions) {

		byte[][] splits = new byte[regions.size() - 1][];

		for (int i = 0; i < regions.size() - 1; i++) {

			splits[i] = regions.get(i).getEndKey().getBytes();

		}

		return splits;

	}

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

		try(final HBaseAdmin admin = clients.getHbaseAdmin(clusterName)) {

			if (!admin.tableExists(table)) {

				createTable(clusterName, table, columns);

			} else {

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

				final Set<String> foundColumns = desc.getFamilies().stream()

						.map(d -> d.getNameAsString())

						.collect(Collectors.toCollection(HashSet::new));

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

				final Collection<String> missingColumns = Sets.difference(columns, 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 {

			for(HBaseRowDescriptor desc : columns) {

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

				d.setWriteToWAL(true);

				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.deleteColumns(Bytes.toBytes(c.getFamily()), Bytes.toBytes(qualifier));

					}

				}

				table.delete(d);

			}

		} 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);

		try(final HTable table = new HTable(conf, tableName)) {

			final ResultScanner rs = table.getScanner(scan);

			try {

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

			} finally {

				rs.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 boolean existHdfsPath(final ClusterName clusterName, final String path) throws HadoopServiceException {

		if (StringUtils.isBlank(path))

			throw new HadoopServiceException("invalid empty 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));

			return hdfs.exists(absolutePath);

		} 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.utils;