D-Net

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

import java.io.ByteArrayInputStream;

import java.io.File;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

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.experimental.runners.Enclosed;

import com.google.common.collect.ArrayListMultimap;

import com.google.common.collect.Multimap;

public class SqoopDriver {

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

    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;

    private String tableMapConf;

    private String enclosed;

    /**

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

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

     * <p/>

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

     * parameters when an new Sqoop Driver instance is created.

     *

     * @param tables

     */

    public void run(Multimap<String, String> tables) throws Exception {

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

            log.info("Importing " + table);

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

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

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

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

                    //			"--input-enclosed-by", enclosed,

                    "--optionally-enclosed-by", enclosed,

                    "--verbose", "--username", dbUser, "--password", dbPass, "--driver", "org.postgresql.Driver", "--batch", "--mapreduce-job-name", "Sqoop Stats Import Job for " + table.getKey(), "--m", sqoopReducersCount};

            int ret = Sqoop.runTool(str);

            if (ret != 0) {

                log.error("Could not run Sqoop Tool " + Integer.toString(ret));

                throw new RuntimeException("Could not run Sqoop Tool " + Integer.toString(ret));

            }

            cleanUp(table.getKey());

        }

    }

    public void initSqoopJob() throws Exception {

        loadTableMappings();

        ArrayList<String> fileNames;

        if (useHdfsStore) {

            fileNames = listHdfsDir();

        } else {

            fileNames = listFilesystemDir();

        }

        // TODO keep this a Multimap collection since we can have multiple files

        // for

        // each table

        // when using multiple reducers ( each reducer creates an output file )

        Multimap<String, String> tables = ArrayListMultimap.create();

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

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

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

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

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

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

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

                }

            }

        }

        long startTime = System.currentTimeMillis();

        log.info("    match   " + tables.entries());

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

    }

    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

    {

        tableMappings = new Properties();

        if (tableMapConf == null) {

            log.error("NULL TABLE MAP CONFIG  IN MAPPER  : ");

        }

        tableMapConf = tableMapConf.replaceAll(",", "\n");

        InputStream stream = new ByteArrayInputStream(tableMapConf.getBytes());

        tableMappings.load(stream);

        stream.close();

        log.info("tm" + tableMappings.entrySet());

        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 Logger getLog() {

        return log;

    }

    public void setLog(Logger log) {

        this.log = log;

    }

    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 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 getConnectionUrl() {

        return connectionUrl;

    }

    public void setConnectionUrl(String connectionUrl) {

        this.connectionUrl = connectionUrl;

    }

    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 String getDelim() {

        return delim;

    }

    public void setDelim(String delim) {

        this.delim = delim;

    }

    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 Configuration getConf() {

        return conf;

    }

    public void setConf(Configuration conf) {

        this.conf = conf;

    }

    public boolean isUseHdfsStore() {

        return useHdfsStore;

    }

    public void setUseHdfsStore(boolean useHdfsStore) {

        this.useHdfsStore = useHdfsStore;

    }

    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;

    }

    public String getTableMapConf() {

        return tableMapConf;

    }

    public void setTableMapConf(String tableMapConf) {

        this.tableMapConf = tableMapConf;

    }

    public String getEnclosed() {

        return enclosed;

    }

    public void setEnclosed(String enclosed) {

        this.enclosed = enclosed;

    }

}