Hbase 数据迁移的常见方式（四）

package com.hbase.test03_etl_with_put_api;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * CompanyEtlWithPutTool
 * @author zhangqingli
 *
 */
public class CompanyEtlWithPutTool {
	private static volatile CompanyEtlWithPutTool INSTANCE;
	public static CompanyEtlWithPutTool getInstance() {
		if (INSTANCE==null) {
			synchronized (CompanyEtlWithPutTool.class) {
				if (INSTANCE==null) {
					INSTANCE = new CompanyEtlWithPutTool();
				}
			}
		}
		return INSTANCE;
	}
	
	/**
	 * connect hbase
	 */
	public HTable getHTable(String tablename) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		return new HTable(conf, tablename);
	}
	public void releaseHTable(HTable htable) {
		try {
			if (htable != null) {
				htable.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * connect mysql
	 */
	public Connection getConn() throws Exception {
		String userName = "root";
		String password = "23wesdxc";
		String url = "jdbc:mysql://dbserver:3306/gsxt";
		Class.forName("com.mysql.jdbc.Driver").newInstance();
		return DriverManager.getConnection(url, userName, password);
	}
	public void releaseConn(Connection conn, Statement st, ResultSet rs) {
		try {
			if (conn != null) {
				conn.close();
			}
			if (st != null) {
				st.close();
			}
			if (rs != null) {
				rs.close();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * extract data from mysql to hbase
	 */
	public void etl() {
		Connection conn = null;
		PreparedStatement pst = null;
		ResultSet rs = null;
		String sql = "select * from company";
		HTable htable = null;
		
		try {
			conn = getConn();
			htable = getHTable("test:company");
			
			pst = conn.prepareStatement(sql);
			rs = pst.executeQuery();
			while (rs.next()) {
				long id = rs.getLong("id");
				String city = rs.getString("city");
				String email = rs.getString("eMail");
				String name = rs.getString("name");
				int num = rs.getInt("num");
				int priority = rs.getInt("priority");
				int state = rs.getInt("state");
				int total_num = rs.getInt("total_num");
				String webAddress = rs.getString("webAddress");
				
				System.out.println(city + " : " + name);
				String rowKey = id+"";
				Put put = new Put(Bytes.toBytes(rowKey));
				byte[] family1 = Bytes.toBytes("info");
				
				if (city!=null) {
					put.add(family1, Bytes.toBytes("city"), Bytes.toBytes(city));
				}
				if (email!=null) {
					put.add(family1, Bytes.toBytes("eMail"), Bytes.toBytes(email));
				}
				if (name!=null) {
					put.add(family1, Bytes.toBytes("name"), Bytes.toBytes(name));
				}
				put.add(family1, Bytes.toBytes("num"), Bytes.toBytes(num));
				put.add(family1, Bytes.toBytes("priority"), Bytes.toBytes(priority));
				put.add(family1, Bytes.toBytes("state"), Bytes.toBytes(state));
				put.add(family1, Bytes.toBytes("total_num"), Bytes.toBytes(total_num));
				if (webAddress!=null) {
					put.add(family1, Bytes.toBytes("webAddress"),
                            Bytes.toBytes(webAddress));
				}
				htable.put(put);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			releaseConn(conn, pst, rs);
			releaseHTable(htable);
		}
	}

	public static void main(String[] args) {
		getInstance().etl();
	}
}