9. Hive使用

作者: 奉先 | 来源:发表于2017-02-28 10:04 被阅读222次

1.Hive数据导入的六种类型：

以下面两个表来实验：

create table emp(
empno int,
ename string,
job string,
mgr int,
hiredate string,
sal double,
comm double,
deptno int)
partitioned by (dt string)
row format delimited
fields terminated by '\t';

create external table dept (
deptno int,
dname string,
loc string)
partitioned by (dt string)
row format delimited 
fields terminated by '\t';

1.从本地文件系统导入到hive表中：

使用load data语句可以直接导入本地文件到hive表中，加关键字local表示从本地系统上传文件到hive表：

hive> load data local inpath '/home/natty.ma/bigdata/hadoop/files/emp.txt' overwrite into table testdb.emp1;

2.从HDFS导入文件到hive表中：

如果不加关键字local表示从hdfs加载文件到hive表。下面先上传本地文件到hdfs，再加载到hive表：

$bin/hdfs dfs -put /home/natty.ma/bigdata/hadoop/files/emp.txt /user/natty.ma
hive> load data inpath '/user/natty.ma/emp.txt' overwrite into table testdb.emp partition (dt='20170228');

load语句实际上是移动文件

3.加载数据覆盖表中已有的数据：

上边2个语句中，load data语句，OVERWRITE参数决定加载文件时是否覆盖。

4.创建表时通过select语句加载：

hive> create table testdb.emp2 as select * from testdb.emp1;

create table ... as 语句会走mapreduce。

5.创建表，通过insert语句加载：

hive> insert into table testdb.emp3 select * from testdb.emp2;

可以增加overwrite选项，来选择insert时，是否覆盖原表的数据。如果不加overwrite参数再执行一次该语句，那么emp3表的数据将会翻倍，查看该表hdfs的目录，会发现有2个文件（而不是一个文件）：

$bin/hdfs dfs -ls /user/hive/warehouse/testdb.db/emp3;
Found 2 items
-rwxrwxrwx   2 root supergroup        661 2017-02-28 10:55 /user/hive/warehouse/testdb.db/emp3/000000_0
-rwxrwxrwx   2 root supergroup        661 2017-02-28 10:57 /user/hive/warehouse/testdb.db/emp3/000000_0_copy_1

6.创建表的时候通过Location指定：

先创建一张表，并指定该表的LOCATION，之后往LOCATON目录（HDFS上的路径）上传数据文件，再查询表时，就可以看到数据了。

create table emp4(
empno int,
ename string,
job string,
mgr int,
hiredate string,
sal double,
comm double,
deptno int)
row format delimited
fields terminated by '\t'
LOCATION '/user/natty.ma/hive_test/emp';

$bin/hdfs dfs -put /home/natty.ma/bigdata/hadoop/files/emp.txt /user/natty.ma/hive_test/emp

PS：我在实验时发现，如果先上传文件，再创建一个表，LOCATION目录是上传的HDFS目录的话，上传的目录会被覆盖清空。

2.Hive导出数据的几种方式：

1.使用Insert语句导出到本地（或HDFS）：

通过LOCAL关键字来确定输出到本地或hdfs。

hive> INSERT OVERWRITE LOCAL DIRECTORY '/home/natty.ma/bigdata/hadoop/output' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' select * from testdb.emp4;
hive> insert overwrite directory '/user/natty.ma/hive_test/output' select * from testdb.emp4;

需要注意的是，在导出到Linux本地目录后，需要cd到上一级目录，再进入该目录才能看到导出的文件。

2.hadoop命令导出：

使用get命令，直接将HDFS上的文件导出到Linux本地系统中。

hive> dfs -get /user/natty.ma/hive_test/emp/* /home/natty.ma/bigdata/hadoop/output/;

3.使用 hive -e "SQL" > xx.txt方式导出文件：

hive -e可以执行执行Hive SQL，将结果导出到文件即可。

hive -e "select * from testdb.emp4" > /home/natty.ma/bigdata/hadoop/output/output.txt

4.使用 sqoop导出文件：

sqoop的使用在后边会介绍。

3.设置 reduce task数目和排序分析：

hive中有4种主要的排序方式：

order by ：全局排序，对所有数据进行排序。
sort by ：局部排序，对每个reduce中的数据进行排序，但是全局不排序。
distribute by：按照某个字段来分区，同一个分区的数据交给一个reduce来处理。
cluster by：是distribute by ... sort by ...的组合，但是不同于上边两项，distribute by 和sort by必须是同一项。

在hive中，可以设置执行作业的reduce的数量：

set mapreduce.job.reduces = <NUMBER>

同时，reduce的数量，也决定了结果输出的文件的个数。

hive> set mapreduce.job.reduces = 3;
hive> insert overwrite local directory '/home/natty.ma/bigdata/hadoop/output/' select * from testdb.emp4 distribute by deptno sort by empno;

上边语句会直接输出3个文件。如果set成5的话，当然会输出5个文件。
我在实验过程中，也发现了一个文件，如果执行的select语句，不需要进行reduce阶段。例如select * 或者select 部分字段，就只会生成一个文件了。

4.设置哪些HQL执行MapReduce：

在执行HQL语句时，我们注意到，有些HQL需要执行MapReduce，而有些则不需要。对于是否执行MapReduce可以进行配置。可以在hive-site.xml或者通过set设置。

<property>
    <name>hive.fetch.task.conversion</name>
    <value>more</value>
    <description>
      Expects one of [none, minimal, more].
      Some select queries can be converted to single FETCH task minimizing latency.
      Currently the query should be single sourced not having any subquery and should not have
      any aggregations or distincts (which incurs RS), lateral views and joins.
      0. none : disable hive.fetch.task.conversion
      1. minimal : SELECT STAR, FILTER on partition columns, LIMIT only
      2. more    : SELECT, FILTER, LIMIT only (support TABLESAMPLE and virtual columns)
    </description>
  </property>

可以有minimal和more来进行选择，首先，可以不进行MR的查询必须是单库查询、不含有任何子查询、不包含聚合函数、distinct、视图、关联等。

1. minimal：

SELECT * ，只筛选partition字段， LIMIT子句。
那么类似的语句不会执行MapReduce：

hive> select * from testdb.emp where dt>='20170227' limit 10;

但是 select empno from testdb.emp where dt>='20170227' limit 10; 会执行MapReduce。
PS：在执行此sql时，报了错误： org.apache.hadoop.hive.ql.ppd.ExprWalkerInfo.getConvertedNode
经查证hive jira，这是hive的一个bug，需要做如下设置，可以避免错误：

hive> set hive.optimize.ppd=false;

2. more ：

如果配置的是more的模式，相比于minimal，如果只select部分字段也不会走MR。

5.Hiveserver2的使用：

Hiverserver（Thrift Hive Server）是老版本的，为了解决同时响应多客户端请求的问题，提供了升级的Hiveserver2版本。

首先，Hive是一个访问HDFS的客户端。但是Hive也是可以实现C/S结构的，Hiveserver2服务开启后，就可以实现Server的功能，可以响应其他（Hive Client端）的请求。那么其他客户端可以连接Hiveserver2服务的Server。

[官方 hiveserver2介绍][1]
[1]:https://cwiki.apache.org/confluence/display/Hive/Setting+Up+HiveServer2

1. 配置hive-site.xml：

启动hiveserver2 服务，需要配置这两项（其余两项默认值即可）：

hive.server2.thrift.port
hive.server2.thrift.bind.host

<property>
    <name>hive.server2.thrift.port</name>
    <value>10000</value>
    <description>Port number of HiveServer2 Thrift interface when hive.server2.transport.mode is 'binary'.</description>
</property>
<property>
    <name>hive.server2.thrift.bind.host</name>
    <value>hadoop-senior01.pmpa.com </value>
    <description>Bind host on which to run the HiveServer2 Thrift service.    </description>
</property>

2. 启动hiveserver2服务：

注意以后台运行方式启动hiveserver2，启动后监控10000端口是否在监听来判断是否启动成功了。

$ hive --service hiveserver2 &
$ netstat -antp | grep 10000

3. 客户端连接hiveserver：

可以以两种方式连接hiveserver： JDBC和Beeline 。
（1）Beeline方式（基于 SQLLine CLI的JDBC客户端）：
下面是以Beeline命令行方式登录：

$ bin/beeline 
beeline> help
beeline> !connect jdbc:hive2://hd-master:10000 root 123456
0: jdbc:hive2://hd-master:10000> show databases;
0: jdbc:hive2://hd-master:10000> select * from testdb.emp4;

beeline的命令在执行时，需要在前边加上!，help命令可以查看具体详细用法。connect连接串的格式 jdbc:hive2://主机名（ip）:端口号用户名密码
用户名和密码是搭建hadoop hive环境的用户名和其密码。

（2）JDBC方式访问：
使用JDBC方式访问hive，classpath需要包含以下jar包:

hadoop-common*.jar
$HIVE_HOME/lib/*.jar

下面一个例子，访问hive中的dept表，并提供了执行更新语句的方式：

package com.pmpa.hiveserver.jdbc;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;

import java.sql.Statement;
import java.util.ArrayList;;

public class HiveJDBC {
    
    private static String jdbcdriver = "org.apache.hive.jdbc.HiveDriver";
    
    public static void main(String[] args) {
        
        ArrayList<Dept> depts = new ArrayList<Dept>();
        
        try {
            Class.forName(jdbcdriver);
        } catch (ClassNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            System.exit(1);
        }
        
        try {
            Connection con = 
                DriverManager.getConnection("jdbc:hive2://localhost:10000/testdb","root", "123456");
            Statement stmt = con.createStatement();
            
            //无返回结果的更新元数据的语句。
            String drop_str = "drop table if exists jdbc_test";
            stmt.execute(drop_str);
            String create_str = "create table jdbc_test (id int, name string, url string)";
            stmt.execute(create_str);
            
            //有返回结果集的查询语句
            String query_str = "select * from dept";
            System.out.println("Running: " + query_str);
            ResultSet rs = stmt.executeQuery(query_str);
            while(rs.next())
            {
                depts.add(new Dept(rs.getInt(1),rs.getString(2)
                        ,rs.getString(3),rs.getString(4)));
            }
            System.out.println("Complete Running: " + query_str);
            
            System.out.print(depts);
            
            
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

其中，Dept是hive表testdb.dept的POJO类。
为了加载classpath，使用下面的shell来配置classpath：

#!/bin/bash
for i in `ls ${HADOOP_HOME}/share/hadoop/common/*.jar`
do
  CLASSPATH=${i}:${CLASSPATH}
done
for i in `ls ${HIVE_HOME}/lib/*.jar`; do
  CLASSPATH=${i}:${CLASSPATH}
done
java -cp ${CLASSPATH}  com.pmpa.hiveserver.jdbc.HiveJDBC

PS：需要注意如果打成jar包，不能使用 java -cp ${classpath} -jar xx.jar xxx 方式，这样的话无法加载classpath。

java -classpath some.jar -jar test.jar
这种方式是不行的，因为使用classpath指定的jar是由AppClassloader来加载，java 命令加了-jar 参数以后，AppClassloader就只关注test.jar范围内的class了，classpath参数失效。

6. Hive UDF开发：

1. UDF介绍

除了Hive中提供的一些基本函数min、max等等外。还可以自行开发定义函数UDF。UDF函数可以直接应用于select语句。需要特别注意以下几点：

自定义UDF需要继承类org.apache.hadoop.hive.ql.exec.UDF
需要实现evaluate()方法，该方法可以重载，不同的重载方法，对应着该函数不同参数用法。
UDF必须要有返回类型，也就是evaluate()方法，可以返回null值，但是该方法不可以是void的。
建议使用hadoop中定义的数据类型，例如Text、LongWritable等，不建议使用java数据类型。

2. UDF开发

下面以一个简单的函数为例，展示Hive UDF的开发过程。
函数功能：输入20170312 返回值 '2017-03-12'。下面是详细步骤：
（1）开发GeneralTest类：

package com.pmpa.hiveserver.udf;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

public class GeneralTest extends UDF {  
    private Text out = new Text();
    public Text evaluate(LongWritable date_int){
        String input = date_int.toString();
        String output = input.substring(0,4) + "-" + input.substring(4,6) + "-" 
                + input.substring(6,8) ;
        out.set(output);
        return out;
    }
}

（2）java程序打成jar包，并发布到目标机器上。
上传到目录：/home/natty.ma/bigdata/hadoop/javaDev/GeneralTest.jar
（3）在hive客户端添加jar包：

hive> add jar /home/natty.ma/bigdata/hadoop/javaDev/GeneralTest.jar;

（4）创建临时函数：

hive> CREATE TEMPORARY FUNCTION general_test_mn AS 'com.pmpa.hiveserver.udf.GeneralTest';

（5）函数测试：

hive> SELECT general_test_mn(20160203);
OK
2016-02-03

3. UDF类型

UDF只能实现“一进一出”操作，如果要“多进一出”需要实现UDAF。所谓一进一出，是输入一行，输出一行，例如上边的例子。多进一出就是所谓的聚合函数。将多行输入记录，聚合输出一行。类似于sum()。

UDF包含以下类型：

UDF： User Defined Function ，一进一出。
UDAF： User Defined Aggregation Function ，多进一出，聚合函数。例如sum，count。
UDTF： User Defined Table Function，一进多出，例如lateral view explore()
下面的文章比较清晰地介绍了这几种UDF的开发方法：

4. UDF开发（继承GenericUDF）

5. UDTF开发：

开发udtf，需要继承类：org.apache.hadoop.hive.ql.udf.generic.GenericUDTF。
为了更好地理解UDTF，可以参考hive内置的一个UDTF，即explode、json_tuple函数等。
需要重载3个方法

// in this method we specify input and output parameters: input ObjectInspector and an output struct
abstract StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException; 

// here we process an input record and write out any resulting records 
abstract void process(Object[] record) throws HiveException;

// this function is Called to notify the UDTF that there are no more rows to process. Clean up code or additional output can be produced here.
abstract void close() throws HiveException;

initialize()方法指定输入和输出参数：输入参数，ObjectInspector；输出参数：struct。
process()方法处理输入记录，并生成结果记录。
close()方法表示没有记录需要处理。