D-Net

package eu.dnetlib.data.mapreduce.hbase.statsExport.drivers;

import java.io.File;

import java.io.FileNotFoundException;

import java.io.IOException;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.Map.Entry;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

import org.apache.log4j.Logger;

import org.apache.sqoop.Sqoop;

import org.junit.Before;

import org.junit.Test;

import eu.dnetlib.data.mapreduce.hbase.statsExport.daos.SqlDAO;

import eu.dnetlib.data.mapreduce.hbase.statsExport.daos.SqlStore;

public class SqoopDriver {

	private Logger log = Logger.getLogger(this.getClass());

	private String TABLE_MAP_PATH = "eu/dnetlib/data/mapreduce/hbase/statsExport/exportTables";

	private Properties tableMappings = new Properties();

	private SqlDAO statsDao;

	private SqlStore statsStore;

	private Properties props;

	private String dbName;

	private String outputPath;

	private String dbUser;

	private String dbPass;

	private String connectionUrl;

	// Number of statements in each batch insert op

	private String RecsPerStatement;

	// Number of statements for each commit ( commit every 1000 inserts, for

	// instance). Set to high value to reduce I/O costs.

	private String StatementPerTrans;

	// Field Seperator ( Delimeter ) used in import HDFS fiels.

	private String delim;

	// Number of reducer classes that Sqoop can use.

	private String sqoopReducersCount;

	private FileSystem hdfs;

	public void init() throws IOException {

		InputStream file = ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/StatsProperties");

		props = new Properties();

		props.load(file);

		file.close();

		this.delim = props.getProperty("Stats.delimCharacter");

		statsStore = new SqlStore();

		statsStore.setDbName(props.getProperty("Stats.dbName"));

		statsStore.setDbDriver(props.getProperty("Stats.dbDriver"));

		statsStore.setDB_URL(props.getProperty("Stats.dbUrl"));

		statsStore.setDbUser(props.getProperty("Stats.dbUser"));

		statsStore.setDbPassword(props.getProperty("Stats.dbPass"));

		statsDao = new SqlDAO();

		statsDao.setStore(statsStore);

		this.outputPath = props.getProperty("Stats.outputPath");

		this.setDbName(props.getProperty("Stats.dbName"));

		this.setConnectionUrl(props.getProperty("Stats.dbUrl") + "/" + props.getProperty("Stats.dbName"));

		this.setDbUser(props.getProperty("Stats.dbUser"));

		this.setDbPass(props.getProperty("Stats.dbPass"));

		this.setReducersCount(props.getProperty("Stats.sqoopReducersCount"));

		this.setRecsPerStatement(props.getProperty("Stats.sqoopRecsPerStatement"));

		this.setStatementPerTrans(props.getProperty("Stats.sqoopStatementPerTrans"));

		hdfs = FileSystem.get(new Configuration());

	}

	/**

	 * Driver for the Sqoop tool. Calls the Sqoop Client for each <input file,

	 * destination table> pair given in the @tables argument.

	 * 

	 * Needed parameters ( connection, database name, etc are set in class

	 * parameters when an new Sqoop Driver instance is created.

	 * 

	 * @param tables

	 */

	public void run(HashMap<String, String> tables) {

		for (Entry<String, String> table : tables.entrySet()) {

			String[] str = { "export", "-Dsqoop.export.records.per.statement =" + RecsPerStatement, "-Dsqoop.export.statements.per.transaction = " + StatementPerTrans,

			"--connect", connectionUrl, "--table", table.getKey(),

			"--export-dir", table.getValue(),

					// TODO : important!! this is necessary since MR multiple

					// outputs writer appends a \t char after writing the KEY

					// field in HDFS.

					// use this so sqoop can ignore it

					"--optionally-enclosed-by", "	", "--input-fields-terminated-by", delim,

					"--verbose", "--username", dbUser, "--password", dbPass, "--batch", "--mapreduce-job-name", "Sqoop Stats Import", "-m", sqoopReducersCount };

			Sqoop.runTool(str);

			cleanUp(table.getKey());

		}

	}

	public void initSqoopJob() throws Exception {

		init(); 

		// PREPARE THE STATS BACKEND ( SHADOW SCHEMA FOR IMPORT)

		prepareDB();

		RemoteIterator<LocatedFileStatus> files;

		try {

			files = hdfs.listFiles(new Path(outputPath), false);

		} catch (FileNotFoundException e1) {

			log.error("HDFS file path with exported data does not exist");

			throw new Exception("HDFS file path with exported data does not exist", e1);

		} catch (IOException e1) {

			log.error("HDFS file path with exported data does not exist");

			throw new Exception("HDFS file path with exported data does not exist", e1);

		}

		HashMap<String, String> tables = new HashMap<String, String>();

		ArrayList<String> fileNames = new ArrayList<String>();

		while (files.hasNext()) {

			String fileName = files.next().getPath().toString();

			fileNames.add(fileName);

		}

		// Table mappings containt the mapping between HDFS files and the Stats

		// DB table that each should be imported to

		for (Entry<Object, Object> e : tableMappings.entrySet()) {

			String name = (String) e.getKey();

			for (String filename : fileNames) {

				String split[] = filename.split("-");

				split[0] = split[0].replaceAll(".*/", "");

				if (split[0].equals(name)) {

					tables.put((String) e.getValue(), filename);

				}

			}

		}

		long startTime = System.currentTimeMillis();

		try {

			this.run(tables);

		} catch (Exception e) {

			log.error("Error while importing tables in Sqoop: ", e);

			throw new Exception("Error while importing tables in Sqoop", e);

		}

		long endtime = System.currentTimeMillis();

		log.info("Time taken for Sqoop Import : " + (endtime - startTime) / 60000 + " minutes");

		finalizedDB();

	}

	/**

	 * Prepares shadow schema in the database that will be used to import data

	 * 

	 * @throws Exception

	 */

	private void prepareDB() throws Exception {

		try {

			statsDao.createSchema(dbName);

		} catch (Exception e) {

			log.error("Error Preparing Schema for DB import. ", e);

			throw new Exception("Error Preparing Schema for DB import.", e);

		}

	}

	private void finalizedDB() throws Exception {

		try {

			statsDao.insertContexts();

			statsDao.buildViews();

			statsDao.executeExtraInserts();

			// Leave indexes out for testing phase

			// statsDao.buildIndexes();

		} catch (Exception e) {

			log.error("Error while finalizing SQL DB", e);

			throw new Exception("Error while finalizing SQL DB", e);

		}

	}

	/**

	 * Cleans up auto-generated Sqoop class files

	 * 

	 * @param table

	 */

	private void cleanUp(String table) {

		try {

			File file = new File(table + ".java");

			if (file.delete()) {

				log.info(file.getName() + " is deleted!");

			} else {

				log.info("Delete operation   failed.");

			}

		} catch (Exception e) {

			log.error("Delete operation   failed" + table);

		}

	}

	public String getConnectionUrl() {

		return connectionUrl;

	}

	public void setConnectionUrl(String connectionUrl) {

		this.connectionUrl = connectionUrl;

	}

	public String getDbUser() {

		return dbUser;

	}

	public void setDbUser(String dbUser) {

		this.dbUser = dbUser;

	}

	public String getDbPass() {

		return dbPass;

	}

	public void setDbPass(String dbPass) {

		this.dbPass = dbPass;

	}

	public String getDelim() {

		return delim;

	}

	public void setDelim(String delim) {

		this.delim = delim;

	}

	public String getReducersCount() {

		return sqoopReducersCount;

	}

	public void setReducersCount(String reducersCount) {

		this.sqoopReducersCount = reducersCount;

	}

	public String getRecsPerStatement() {

		return RecsPerStatement;

	}

	public void setRecsPerStatement(String recsPerStatement) {

		RecsPerStatement = recsPerStatement;

	}

	public String getStatementPerTrans() {

		return StatementPerTrans;

	}

	public void setStatementPerTrans(String statementPerTrans) {

		StatementPerTrans = statementPerTrans;

	}

	public Logger getLog() {

		return log;

	}

	public void setLog(Logger log) {

		this.log = log;

	}

	public String getTABLE_MAP_PATH() {

		return TABLE_MAP_PATH;

	}

	public void setTABLE_MAP_PATH(String tABLE_MAP_PATH) {

		TABLE_MAP_PATH = tABLE_MAP_PATH;

	}

	public Properties getTableMappings() {

		return tableMappings;

	}

	public void setTableMappings(Properties tableMappings) {

		this.tableMappings = tableMappings;

	}

	public SqlDAO getStatsDao() {

		return statsDao;

	}

	public void setStatsDao(SqlDAO statsDao) {

		this.statsDao = statsDao;

	}

	public SqlStore getStatsStore() {

		return statsStore;

	}

	public void setStatsStore(SqlStore statsStore) {

		this.statsStore = statsStore;

	}

	public Properties getProps() {

		return props;

	}

	public void setProps(Properties props) {

		this.props = props;

	}

	public String getDbName() {

		return dbName;

	}

	public void setDbName(String dbName) {

		this.dbName = dbName;

	}

	public String getOutputPath() {

		return outputPath;

	}

	public void setOutputPath(String outputPath) {

		this.outputPath = outputPath;

	}

	public String getSqoopReducersCount() {

		return sqoopReducersCount;

	}

	public void setSqoopReducersCount(String sqoopReducersCount) {

		this.sqoopReducersCount = sqoopReducersCount;

	}

	public FileSystem getHdfs() {

		return hdfs;

	}

	public void setHdfs(FileSystem hdfs) {

		this.hdfs = hdfs;

	}

}