大数据案例——Hive、MySQL、HBase数据互导

本文是基于教程大数据案例——本地文件上传hive数据库，对数据在Hive，MySQL，HBase之间的互导进行进阶版的学习。数据集：淘宝用户点击商品操作数据集，共30万条记录。如果没有安装MySQL，HBase，Sqoop请参考博文安装：Ubuntu安装MySQL及常用操作，Ubuntu安装HBase及常用操作，Ubuntu安装Sqoop1。

一、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]));// rowkeyfor(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");}}