D-Net

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

import java.io.File;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.Map.Entry;

import java.util.Properties;

import org.apache.commons.cli.CommandLine;

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

import org.apache.log4j.Logger;

import org.apache.sqoop.tool.SqoopTool;

import com.cloudera.sqoop.SqoopOptions;

import com.cloudera.sqoop.mapreduce.ExportOutputFormat;

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 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 fields.

	private String delim;

	// Number of reducer classes that Sqoop can use.

	private String sqoopReducersCount;

	private FileSystem hdfs;

	private Configuration conf;

	private boolean useHdfsStore = true;

	private boolean batch = true;

	private boolean verbose = true;

	/**

	 * 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) throws Exception {

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

			log.info("Importing " + table);

			SqoopTool tool = SqoopTool.getTool("export");

			if (null == tool) {

				log.error("No such sqoop tool: " + "dnetSqoop" + ". See 'sqoop help'.");

				throw new RuntimeException("Could not get Sqoop  Export Tool ");

			}

			SqoopOptions opts = new SqoopOptions();

			opts.setConnectString(connectionUrl);

			opts.setExportDir(table.getValue());

			opts.setTableName(table.getKey());

			opts.setUsername(dbUser);

			opts.setPassword(dbPass);

			opts.setInputFieldsTerminatedBy(delim.charAt(0));

			opts.setBatchMode(batch);

			opts.setVerbose(verbose);

			opts.setNumMappers(Integer.parseInt(sqoopReducersCount));

			// 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

			opts.setInputEnclosedBy('	');

			opts.setInputEncloseRequired(false);

			opts.setExportDir(table.getValue());

			opts.setTableName(table.getKey());

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

			  args.add("-Dsqoop.statements.per.transaction");

			  args.add("10000"); 

			  args.add("-Dsqoop.statements.per.transaction");

			  args.add("10000"); 

//			  CommonArgs.addHadoopFlags(args);

			  args.add(ExportOutputFormat.RECORDS_PER_STATEMENT_KEY + "="

			           + "1000");

			       args.add("-D");

			      args.add(ExportOutputFormat.STATEMENTS_PER_TRANSACTION_KEY + "="

		           + "1000"); 

			opts.setExtraArgs( args.toArray(new String[0]));

			conf.set("-Dsqoop.statements.per.transaction", "1000");

			conf.set("-Dsqoop.records.per.statement", "1000");

			opts.setConf(conf);

			tool.appendArgs( args.toArray(new String[0]));

//			

			tool.run(opts);

			cleanUp(table.getKey());

		}

	}

	public static void main(String[] args) {

		SqoopDriver sqoopDriver = new SqoopDriver();

		BasicConfigurator.configure();

		sqoopDriver.setDelim("!");

		sqoopDriver.setOutputPath("/tmp/test_stats/");

		sqoopDriver.setConnectionUrl("jdbc:postgresql://duffy.di.uoa.gr:5432/test_stats");

		sqoopDriver.setDbUser("sqoop");

		sqoopDriver.setDbPass("sqoop");

		sqoopDriver.setReducersCount("1");

		sqoopDriver.setRecsPerStatement("1000");

		sqoopDriver.setStatementPerTrans("1000");

		sqoopDriver.setUseHdfsStore(true);

		try {

			sqoopDriver.initSqoopJob();

		} catch (Exception e) {

			System.out.print("ERROR " + e.toString());

			e.printStackTrace();

		}

	}

	public void initSqoopJob() throws Exception {

		loadTableMappings();

		ArrayList<String> fileNames;

		if (useHdfsStore) {

			fileNames = listHdfsDir();

		} else {

			fileNames = listFilesystemDir();

		}

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

		// 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();

			log.info("Found tableMappings   " + name);

			for (String filename : fileNames) {

				log.info("  filename    " + filename);

				String str = filename.substring(filename.lastIndexOf('/') + 1);

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

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

				log.info("  filename processed  " + split[0]);

				log.info(split[0]);

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

					log.info("    match   " + e.getValue() + " " + filename);

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

				}

			}

		}

		long startTime = System.currentTimeMillis();

		try {

			log.info("Running for" + tables);

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

	}

	public ArrayList<String> listHdfsDir() throws Exception {

		if (conf == null) {

			conf = new Configuration();

		}

		hdfs = FileSystem.get(conf);

		RemoteIterator<LocatedFileStatus> Files;

		try {

			Path exportPath = new Path(hdfs.getUri() + outputPath);

			Files = hdfs.listFiles(exportPath, false);

			if (Files == null) {

				log.error("No files generated in " + outputPath + " at " + exportPath.getName());

				throw new Exception("No files generated in " + outputPath + " at " + exportPath.getName() + " in " + hdfs.resolvePath(exportPath));

			}

		} catch (Exception e) {

			log.error("HDFS file path with exported data does not exist : " + outputPath + " at " + new Path(hdfs.getUri() + outputPath));

			throw new Exception("HDFS file path with exported data does not exist :   " + outputPath + " at " + new Path(hdfs.getUri() + outputPath), e);

		}

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

		while (Files.hasNext()) {

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

			log.info("Found hdfs file " + fileName);

			fileNames.add(fileName);

		}

		return fileNames;

	}

	public ArrayList<String> listFilesystemDir() throws Exception {

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

		String files;

		File folder = new File(outputPath);

		File[] listOfFiles = folder.listFiles();

		if (listOfFiles == null) {

			log.error("No files generated in " + outputPath);

			throw new Exception("No files generated in " + outputPath);

		}

		for (int i = 0; i < listOfFiles.length; i++) {

			if (listOfFiles[i].isFile()) {

				files = listOfFiles[i].getAbsolutePath();

				log.info("Found " + files);

				fileNames.add(files);

			}

		}

		return fileNames;

	}

	public void loadTableMappings() throws Exception

	{

		InputStream file = ClassLoader.getSystemResourceAsStream(TABLE_MAP_PATH);

		tableMappings.load(file);

		file.close();

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

			throw new Exception("Could not load Table Mconfappings in sqoop init job");

		}

	}

	/**

	 * 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 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;

	}

	public boolean isUseHdfsStore() {

		return useHdfsStore;

	}

	public void setUseHdfsStore(boolean useHdfsStore) {

		this.useHdfsStore = useHdfsStore;

	}

	public Configuration getConf() {

		return conf;

	}

	public void setConf(Configuration conf) {

		this.conf = conf;

	}

	public boolean isBatch() {

		return batch;

	}

	public void setBatch(boolean batch) {

		this.batch = batch;

	}

	public boolean isVerbose() {

		return verbose;

	}

	public void setVerbose(boolean verbose) {

		this.verbose = verbose;

	}

}