import java.io.UnsupportedEncodingException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hive.cli.CliSessionState;
import org.apache.hadoop.hive.common.io.SessionStream;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.Driver;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Main extends Configured implements Tool {
	public static void main(String[] args) throws Exception {
		if (args.length != 3) {
			System.out.println("Usage: <source HDFS path> <target database> <target table>");
			System.exit(-1);
		}
		int res = ToolRunner.run(new Configuration(), new Main(), args);
		System.exit(res);
	}

	@Override
	public int run(String[] arg0) throws Exception {
		String hdfsPath = arg0[0];
		String databaseName = arg0[1];
		String tableName = arg0[2];

		Configuration conf = getConf();
		HiveConf hiveConf = new HiveConf(conf, Main.class);
		loadDataintoHiveTable(conf, hiveConf, hdfsPath, databaseName, tableName);
		return 0;
	}

	public static void loadDataintoHiveTable(Configuration configuration, HiveConf hiveConf, String hdfsPath,
			String databaseName, String tableName) throws Exception {
		CliSessionState ss = new CliSessionState(hiveConf);
		ss.in = System.in;
		try {
			ss.out = new SessionStream(System.out, true, "UTF-8");
			ss.err = new SessionStream(System.err, true, "UTF-8");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}

		SessionState.start(ss);
		Driver driver = new Driver(hiveConf);
		try {
			String cmd = "Load data inpath '" + hdfsPath + "' into table " + databaseName + "." + tableName;
			System.out.println("**** THE HQL STATEMENT IS :    " + cmd);
			driver.run(cmd);
		} catch (Exception e) {
			throw e;
		} finally {
			if (ss != null) {
				ss.close();
				ss = null;
			}
			if (driver != null) {
				driver.close();
				driver = null;
			}
		}
	}
}