一、Hive預操作

1、建立臨時表user_action

hive> create table dblab.user_action(id STRING,uid STRING, item_id STRING, behavior_type STRING, item_category STRING, date DATE, province STRING) COMMENT 'Welcome to xmu dblab! ' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE;

2、將small_user表中的資料插入到user_action(執行時間：10秒左右)

hive> INSERT OVERWRITE TABLE dblab.user_action select * from dblab.small_user;

查詢命令是否成功插入。select * from user_action limit 10;

二、使用Sqoop將資料從Hive匯入MySQL

1、啟動hadoop叢集、MySQL服務

start-all.sh
service mysql start

2、將前面生成的臨時表資料從 HDFS 匯入到 MySQL 中，步驟如下：(1)~(4)操作都是在 MySQL 互動客戶端執行。
(1)登入 MySQL，回車並輸入密碼

mysql –u hive –p 

(2)建立資料庫

mysql> show databases; #顯示所有資料庫
mysql> create database dblab; #建立dblab資料庫
mysql> use dblab; #使用資料庫

注意：檢視資料庫的編碼show variables like "char%";,請確認當前編碼為utf8，否則無法匯入中文，請參考Ubuntu安裝MySQL及常用操作修改編碼。
(3)建立表，並設定其編碼為utf-8

mysql> CREATE TABLE `dblab`.`user_action` 
 
(`id` varchar(50),`uid` varchar(50),`item_id` varchar(50),`behavior_type` varchar(10),`item_category` varchar(50), `date` DATE,`province` varchar(20)) ENGINE=InnoDB DEFAULT CHARSET=utf8;

提示：語句中的引號是反引號`，不是單引號’。
建立成功後，退出 MySQL。
(4)匯入資料(執行時間：20秒左右)

cd /usr/local/sqoop #進入 sqoop 安裝主目錄
bin/sqoop export --connect jdbc:mysql://localhost:3306/dblab --username hive --password hive --table user_action --export-dir '/user/hive/warehouse/dblab.db/user_action' --fields-terminated-by '\t'; #匯入命令

注意：IP 部分需要使用 HadoopMaster 節點對應的 IP 地址。
欄位解釋：

bin/sqoop export  ##表示資料從 hive 複製到 mysql 中
--connect jdbc:mysql://localhost:3306/dblab 
--username hive  #mysql登陸使用者名稱
--password hive  #登入密碼
--table user_action  #mysql 中的表，即將被匯入的表名稱  
--export-dir '/user/hive/warehouse/dblab.db/user_action '  #hive 中被匯出的檔案 
--fields-terminated-by '\t'   #hive 中被匯出的檔案欄位的分隔符

3、檢視MySQL中user_action表資料。

三、使用Sqoop將資料從MySQL匯入HBase

1、啟動hadoop叢集、MySQL服務、HBase服務

start-all.sh
service mysql start
start-hbase.sh

2、登陸HBase shell

hbase shell

3、建立表user_action

hbase> create 'user_action', { NAME => 'f1', VERSIONS => 5}

4、匯入資料(執行時間：30秒左右)

sqoop  import  --connect jdbc:mysql://localhost:3306/dblab --username hive --password hive --table user_action --hbase-table user_action --column-family f1 --hbase-row-key id --hbase-create-table -m 1

注意：IP部分改為本機IP地址或localhost。同時，HBase只支援十六進位制儲存中文。
欄位解釋：

bin/sqoop  import  --connect  jdbc:mysql://localhost:3306/dblab
--username  hive     
--password  hive 
--table user_action
--hbase-table user_action #HBase中表名稱
--column-family f1 #列簇名稱
--hbase-row-key id #HBase 行鍵
--hbase-create-table #是否在不存在情況下建立表
-m 1 #啟動 Map 數量

5、檢視HBase中user_action表資料

四、HBase Java API訪問統計資料

1、啟動hadoop叢集、HBase服務

start-all.sh
start-hbase.sh

2、資料準備
將之前的 user_action 資料從 HDFS 複製到本地。

cd ~/dblab
hdfs dfs -get /user/hive/warehouse/dblab.db/user_action .  #將HDFS上的user_action資料複製到本地當前目前，注意'.'表示當前目錄
cd user_action
cat 00000* > user_action.output #將00000*檔案複製一份重新命名為user_action.output，*表示萬用字元
head user_action.output  #檢視user_action.output前10行

3、Eclipse編寫ImportHBase程式，並打包成可執行jar包，命名為ImportHBase.jar，儲存至~/dblab/。java程式碼附在文末。
4、資料匯入(執行時間：2分鐘左右)
使用 Java 程式將資料從本地匯入 HBase 中，匯入前請先清空user_action表truncate 'user_action'：

hadoop jar ~/dblab/ImportHBase.jar com.dblab.hbase.HBaseImportTest /home/dblab/dblab/user_action/user_action.output

欄位解釋：

hadoop jar  #hadoop jar包執行方式
~/ImportHBase.jar  #jar包的路徑
com.dblab.hbase.HBaseImportTest   #主函式入口 
/home/dblab/dblab/user_action/user_action.output  #main方法接收的引數args

5、檢視HBase中user_action表資料

附錄：ImportHBase.java

package com.dblab.hbase;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseImportTest extends Thread {
    public Configuration config;
    public HTable table;
    public HBaseAdmin admin;

    public HBaseImportTest() {
        config = HBaseConfiguration.create();
//      config.set("hbase.master", "master:60000");
//      config.set("hbase.zookeeper.quorum", "master");
        try {
            table = new HTable(config, Bytes.toBytes("user_action"));
            admin = new HBaseAdmin(config);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        if (args.length == 0) {       //第一個引數是該jar所使用的類，第二個引數是路徑
            throw new Exception("You must set input path!");
        }

        String fileName = args[args.length-1];  //輸入的檔案路徑是最後一個引數
        HBaseImportTest test = new HBaseImportTest();
        test.importLocalFileToHBase(fileName);
    }

    public void importLocalFileToHBase(String fileName) {
        long st = System.currentTimeMillis();
        BufferedReader br = null;
        try {
            br = new BufferedReader(new InputStreamReader(new FileInputStream(
                    fileName)));
            String line = null;
            int count = 0;
            while ((line = br.readLine()) != null) {
                count++;
                put(line);
                if (count % 10000 == 0)
                    System.out.println(count);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {

            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            try {
                table.flushCommits();
                table.close(); // must close the client
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
        long en2 = System.currentTimeMillis();
        System.out.println("Total Time: " + (en2 - st) + " ms");
    }

    @SuppressWarnings("deprecation")
    public void put(String line) throws IOException {
        String[] arr = line.split("\t", -1);
        String[] column = {"id","uid","item_id","behavior_type","item_category","date","province"};

        if (arr.length == 7) {
            Put put = new Put(Bytes.toBytes(arr[0]));// rowkey
            for(int i=1;i<arr.length;i++){
                put.add(Bytes.toBytes("f1"), Bytes.toBytes(column[i]),Bytes.toBytes(arr[i]));
            }
            table.put(put); // put to server
        }
    }

    public void get(String rowkey, String columnFamily, String column,
            int versions) throws IOException {
        long st = System.currentTimeMillis();

        Get get = new Get(Bytes.toBytes(rowkey));
        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));

        Scan scanner = new Scan(get);
        scanner.setMaxVersions(versions);

        ResultScanner rsScanner = table.getScanner(scanner);

        for (Result result : rsScanner) {
            final List<KeyValue> list = result.list();
            for (final KeyValue kv : list) {
                System.out.println(Bytes.toStringBinary(kv.getValue()) + "\t"
                        + kv.getTimestamp()); // mid + time
            }

        }
        rsScanner.close();

        long en2 = System.currentTimeMillis();
        System.out.println("Total Time: " + (en2 - st) + " ms");
    }

}