BG.Hive - part3

Posted 2020-09-04 Niko12230

1. Hive数据导入 - Load

　　Load，加载，不会进行任何数据转换和处理的操作，只会进行数据Move操作，将元数据移动到HDFS指定目录

　　a> hdfs dfs -put ‘hdfs的hive目录‘

　　b> LOAD DATA [LOCAL] INPATH ‘filepath‘ [OVERWRITE] INTO TABLE tablename [PARTITION(partcol1=val1,partcol2=val2...)]

　　　　filepath: 要加载的数据文件路径，可以是相对路径，也可以是绝对路径，或者hdfs url；filepath为文件则移动文件，如果是目录，则将该目录下所有的文件移动到目标目录（表/分区）

　　　　LOCAL: 要加载的文件为本地文件 (file:///user/hive/...)

　　　　OVERWRITE: 目标目录（表/分区）下的数据会全部删除，然后覆盖；没有改关键字，则是增加的模式；

　　vim load.txt　　#创建本地文件， 数据如下

1,Niko,Shanghai,2016-01-01
2,Jim,Beijing,2017-01-01
3,Ye,GuangZhou,2018-01-01

　　hive> create table loaddata(id int, name string, city string, createdate date)　　#在Hive中创建表
    > row format delimited
    > fields terminated by ‘,‘
    > stored as textfile;

　　hive> load data local inpath ‘/root/load.txt‘ into table loaddata;　　#Load本地文件到表中

　　hive> load data inpath ‘file:///root/load.txt‘ into table loaddata;　　#使用权全路径加载本地文件

　　hive> select * from loaddata;　　#查询loaddata表，数据已经导入，由于前面2次都没有使用overwrite，可以发现，表中有6条数据

　　hive> load data local inpath ‘/root/load.txt‘ overwrite into table loaddata;　　#通过overwrite加载数据，成功后会发现loaddata表只有3条数据。之前的数据被清除了

　　hive> truncate table loaddata;　　#清空表

　　hdfs dfs -put load.txt ‘/user/hive/‘

　　hive> load data inpath ‘/user/hive/load.txt‘ into table loaddata;　　#由hdfs加载数据到loaddata表

　　hive> load data inpath ‘hdfs://bigdata.hadoop.master:9000/user/hive/load.txt‘ into table loaddata;　　#使用hdfs全uri指定文件进行导入

　　

　　hdfs dfs -mkdir /user/hive/loadfolder　　#在hdfs中创建目录

　　hdfs dfs -cp file:///root/load.txt ‘/user/hive/loadfolder/load1.txt‘　　#将本地文件复制到hdfs

　　hdfs dfs -cp file:///root/load.txt ‘/user/hive/loadfolder/load2.txt‘　　#将本地文件复制到hdfs，生成2份

　　hive> load data inpath ‘/user/hive/loadfolder/‘ into table loaddata;　　#将hdfs目录作为路径到如Hive，会发现该目录下的2个文件数据都已经导入了Hive表中。并在Hive的目录下生成了2个文件。

　　

2. Hive数据导入 - Insert

　　INSERT OVERWRITE　　#覆盖目标表，immutable不限制

　　INSERT INTO　　#向目标表appent，immutable限制

　　hive> create table insertdata like loaddata;　　#创建表

　　hive> alter table insertdata set tblproperties (‘immutable‘=‘true‘);　　#将insertable的数据设置为不可变

　　　　immutable，当设置为true时，表中存在数据，不允许在修改数据；表中没有数据时，可以新增数据

　　hive> insert into insertdata select * from loaddata;　　#在insertdata没有数据的情况下，该语句执行成功

　　hive> insert into insertdata select * from loaddata;　　#再次执行，FAILED: SemanticException [Error 10256]: Inserting into a non-empty immutable table is not allowed insertdata

　　hive> insert overwrite table insertdata select * from loaddata;　　#Overwrite，在inserttable中有数据的情况下也能执行

　　hive> from loaddata insert overwrite table insertdata select *;　　#from在前面，overwrite

　　hive> from loaddata insert into insertdata select *;　　#from在前面，into；需要先关闭immutable

　　hive> create table insertdata2 like loaddata;　　#在创建一个表

　　hive> from loaddata　　#使用from同时向多张表插入数据
    > insert overwrite table insertdata select *
    > insert into table insertdata2 select *;

　　

3. Hive数据导出

　　导出到本地文件系统

　　导出到HDFS

　　Hive -e 或 -f，导入到其他表

　　hive> insert overwrite local directory ‘/root/loadexport‘ select * from loaddata;　　#将loaddata表数据导入到本地文件

　　hive> insert overwrite directory ‘/user/hive/loadexport‘ select * from loaddata;　　#导出到HDFS，字段分隔符为^A

　　hive -e ‘select * from loaddata‘;　　#将查询结果进行显示

　　hive -e ‘select * from loaddata‘ >> loaddatae.txt;　　#通过>>管道符，将查询结果写入loaddata.txt文件；>>代表追加；>代表覆盖；字段间分割符为Tab

　　vim loaddata_query.hql　　#编写查询语句的文件，内容为　　use hellohive　　(\n)　　select * from loaddata;

　　hive -f loaddata_query.hql > loaddatae.txt;　　#通过hive-f的方式，将查询结果覆盖到loaddatae.txt文件

 

4. Hive复杂类型

　　数组，ARRAY<date_type>：有顺序的并且数组元素都必须是相同的数据类型，由0开始索引

　　键值对，MAP<primitive_type,data_type>：例如map<int,string>

　　结构体，STRUCT<col_name1:data_type,col_name2:data_type...>

　　联合体，UNIONTYPE<data_type,data_type...>

　　fields terminated by: 字段间分隔符（默认\001 ^A）；collection items terminated by: 集合元素间分隔符（默认\002 ^B）；map keys terminated by: map键值间分隔符（默认\003 ^C）；lines terminated by: 行分隔符（默认\n）

hive> create external table if not exists vendor(　　　　　　#创建表--复杂数据类型
> id int comment ‘供应商ID‘,
> name string comment ‘供应商名称‘,
> contacter array<string> comment ‘联系人‘,
> productline map<string,double> comment ‘产品线和产品价格‘,
> address struct<country:string,city:string,zip:int> comment ‘地址（国家，城市，邮编）‘
> )
> row format delimited
> fields terminated by ‘,‘
> collection items terminated by ‘ ‘
> map keys terminated by ‘:‘
> lines terminated by ‘\n‘
> stored as textfile;

 

　　创建数据文件，vendorrecord.txt

1,VendorA,Name1 Name2 Name3,P1:100.0 P2:200.0 P3:50,CN Shanghai 200000
2,VendorB,Name1 Name2 Name3,P1:1000 P2:20 P3:80,CN Beijing 100000
3,VendorC,Name100 Name200 Name300,P100:3 P200 33.8,CN Shanghai 200000

 

 

　　导入数据到vendor表

　　hive> load data local inpath ‘/root/vendorrecord.txt‘ into table vendor;

　　hive> select * from vendor;　　#查询导入的数据，结果如下

 

OK
1 VendorA ["Name1","Name2","Name3"] {"P1":100.0,"P2":200.0,"P3":50.0} {"country":"CN","city":"Shanghai","zip":200000}
2 VendorB ["Name1","Name2","Name3"] {"P1":1000.0,"P2":20.0,"P3":80.0} {"country":"CN","city":"Beijing","zip":100000}
3 VendorC ["Name100","Name200","Name300"] {"P100":3.0,"P200":null,"33.8":null} {"country":"CN","city":"Shanghai","zip":200000}
Time taken: 1.752 seconds, Fetched: 3 row(s)

 

　　hive> select contacter from vendor;　　#查询联系人　　数组

　　hive> select contacter[0] from vendor;　　#查询每条记录的第一个联系人　　数组下标　　数组

　　hive> select size(contacter) from vendor;　　#查询每条记录的联系人个数　　数据元素个数　　数组

　　hive> select array_contains(contacter,‘Name1‘) from vendor;　　#查询每条记录的联系人是否包含某个名称　　判断数组是否包含某个元素，返回值为true,false　　数组

　　hive> select sort_array(contacter) from vendor;　　#查询每条记录时对联系人进行排序　　数组排序　　数组

　　hive> select productline from vendor;　　#查看产品线

　　hive> select productline[‘P1‘] from vendor;　　#查看某个产品的所有价格　　MAP根据Key查询Value　　MAP

　　hive> select size(productline) from vendor;　　#查看产品个数　　MAP键值对个数　　MAP

　　hive> select map_keys(productline) from vendor;　　#查看所有产品　　查询MAP的所有key　　MAP

　　hive> select map_values(productline) from vendor;　　#查看所有产品价值　　查询MAP的所有Value　　MAP

　　hive> select address.country from vendor;　　#查看所有记录所在的国家　　查看STRUCT的某个属性　　STRUCT

 

5. Hive常见优化

　　小文件合并

　　小文件过多，导致HDFS的NameNode负载过大；每个HDFS文件都在NameNode中会有对应的记录，NameNode是直接于在与内存中的

　　小文件过多，导致Metastore解析文件负担过大；

　　建议每次执行完一个SQL后就进行小文件合并；

　　扫描文件 -- 列剪裁

　　控制参数： hive.optimize.cp=true;

　　使用select具体的列代替select *

　　扫描文件 -- 分区剪裁

　　控制参数： hive.optimize.pruner=true;

　　在查询子句中使用分区字段当作查询条件

　　Join优化 -- 控制启动的Job个数

　　无论多少张表Join，如果Join的Key都一样，则都会合并为一个MR作业

　　Join优化 -- 左边表为小表

　　在Reduce阶段，Join左边的表会全部加载到内存；此原则不适用于mapjoin

　　Join优化 -- MapJoin

　　Join放生在Map端，无需Reduce任务；MapJoin绝对不存在数据倾斜；Map Join要求查询的表很小（通常小于100M）

　　select /*+mapjoin(a)*/ a.id,b.name from TableA a join TableB b on a.id=b.id

　　Group by优化

　　控制参数： hive.groupby.skewindata=true;　　默认为false

　　主要优化数据倾斜问题

　　控制参数设置为true后，查询语句的执行计划会生成2个MR作业，第一个为预聚合，第二个是根据预聚合的结果进行最终聚合

　　空值：大量空值容易导致数据倾斜 => 数据预处理

 

6. UDFs

　　UDFs : User Defined Functions　　Inupt 1 Output 1

　　UDAF : User Defined Aggregation Function　　Input N Output 1

　　UDTF : User Defined Table- Generating Function　　Inpurt 1 Output N

　　内置UDF： string, Data ....

　　内置UDAF：Count, Sum, AVG ...

　　内置UDTF： Explode ...

　　自定义UDFs：https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF

　　show functions;　　desc function year;　　desc function extended year;

　　UDF源码，insight.io　　如（+）源码：https://www.insight.io/github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDFOPPlus.java

　　Group by 操作源码：https://www.insight.io/github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByOperator.java

　　a> 编写UDF函数的程序，并打成Jar包　　b> 在Hive中使用add jar的方式安装该Jar包　　c> 在Hive中调用该函数

　　编写UDF函数，输入身份证号，输出年龄

　　在ItelliJ IDEA中编写函数，代码如下；并生成Jar包（Project Structure => Artifacts => Add...）　　

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;
import java.util.Calendar;
import java.lang.*;

public class getAgeFromId extends UDF {
    Calendar cal = Calendar.getInstance();
    int year = cal.get(Calendar.YEAR);

    public Text evaluate (Text s) throws Exception
    {
        if(s == null){return  null;}
        if(s.getLength() != 18)
        {
            throw new Exception("Exactly one argument is expected.");
        }
        int age = year - Integer.parseInt(s.toString().substring(6,10));
        return new Text(Integer.toString(age));
    }
}

 

　　hive> add jar /root/Hive_UDF.jar　　#在Hive中导入Jar包

　　hive> create temporary function getAge as ‘getAgeFromId‘;　　#根据自定义的函数在Hive中创建临时函数

　　hive> select getage(‘111222199001013333‘);　　#测试自定义函数

　　上面由于时临时函数，当Hive重启后，该函数将不可用

　　解决方案：

　　vim /opt/hive-2.1.1/conf/.hiverc　　#将下面的语句写入hiverc文件； hiverc隐藏文件在每次Hive启动时，里面的每条语句都会被重新执行

　　　　add jar /root/Hive_UDF.jar

　　　　create temporary function getAge as ‘getAgeFromId‘;