package hbase;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.conf.Configuration;

import com.cloudera.sqoop.SqoopOptions;
import com.cloudera.sqoop.tool.ImportTool;

public class hbase {
	
	public static void execsqoop(String stable,	String htable,String cf,
			List<String> columnamesonly,String colurl,String coldri,
			String colusr,String colpass)
	{
		try
		{
			System.out.println("the hbase table name is:"+htable);
			System.out.println("the Subtable name is:"+stable);
			System.out.println("the column family name is:"+cf);
			System.out.println("the columnname is:"+columnamesonly);
			System.out.println("************##############************");
			Configuration conf=new Configuration();
			conf.set("hbase.zookeeper.quorum", "zk1.alp.com,zk2.alp.com,zk3.alp.com");
			String JOB_NAME = "Sqoop HBase Job";
			String MAPREDUCE_JOB = "HBase Map Reduce Job";
			String sqURL = colurl;
			String sqDRIVER = coldri;
			String sqUSERNAME = colusr;
			String sqPASSWORD = colpass;
			String HADOOP_HOME = "/var/lib/hadoop-hdfs";
			String JAR_OUTPUT_DIR = "/home/hbase/tmp/";
			String JAR_OUTPUT_DIR1 = "/home/sqoopjar";
			String SUCCESS = "SUCCESS !!!";
			String FAIL = "FAIL !!!";
			String arraycol[]=new String[columnamesonly.size()];
			for (int j=0;j<columnamesonly.size();j++)
			{
				arraycol[j]=columnamesonly.get(j);
			}			    
			int retVal = 0;
			SqoopOptions options = new SqoopOptions(sqURL,stable);
			options.setDriverClassName(sqDRIVER);
			options.setConf(conf);
			options.setUsername(sqUSERNAME);
			options.setPassword(sqPASSWORD);
			options.setHBaseTable(htable);
			options.setHBaseRowKeyColumn("rowkey");
			options.setHBaseColFamily(cf);
			options.setSplitByCol("rowkey");
			options.setJobName(JOB_NAME);
			options.setColumns(arraycol);
			//options.setJobName(MAPREDUCE_JOB);
			options.setTableName(stable);
			options.setJarOutputDir(JAR_OUTPUT_DIR);
			options.setJarOutputDir(JAR_OUTPUT_DIR1);
			//options.setCreateHBaseTable(true);		    
			options.setDirectMode(true);
			ImportTool it = new ImportTool();
			//it.validateOptions(options);
			//retVal = it.run(options);
			it.wait(5000);
			if(retVal == 0)
			{
			    System.out.println(SUCCESS);
			}
			else
			{
			  	System.out.println(FAIL);
			}
			System.out.println("the hbase table name is:"+htable);
			System.out.println("the Subtable name is:"+stable);
			System.out.println("the column family name is:"+cf);
			System.out.println("the columnname is:"+Arrays.toString(arraycol));
			conf.clear();
				
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}    
	}
	public static void main(String[] args){
		
		hbase h=new hbase();
		String stable ="t2_stable1";
		String htable="tab2";
		String cf="cf1";
		String cf2="cf2";
		ArrayList<String> columnamesonly=new ArrayList<String>();
		ArrayList<String> columnamesonly1=new ArrayList<String>();
		columnamesonly.add("t2_col1");
		columnamesonly.add("t2_col2");
		columnamesonly.add("t2_col4");
		columnamesonly.add("t2_col5");columnamesonly.add("rowkey");
		
		columnamesonly1.add("t2_col3");columnamesonly1.add("t2_col6");columnamesonly1.add("rowkey");
		String colurl="jdbc:mysql://168.129.1.309:3306/stagging";
		String coldri="com.mysql.jdbc.Driver";
		String colusr="root";
		String colpass="root";
		
		hbase.execsqoop(stable, htable, cf2, columnamesonly1, colurl, coldri, colusr, colpass);
		hbase.execsqoop(stable, htable, cf, columnamesonly, colurl, coldri, colusr, colpass);
	}


}