ODS建表和数据加载


ODS层的数据与源数据的格式基本相同。

创建ODS层表:

1
2
3
4
5
6
7
8
9
10
use ODS;

create external table ods.ods_start_log( `str` string)
comment '用户启动日志信息'
partitioned by (`dt` string)
location '/user/data/logs/start';

-- 加载数据的功能(测试时使用)
alter table ods.ods_start_log add partition(dt='2020-08-02');
alter table ods.ods_start_log drop partition (dt='2020-08- 02');

加载启动日志数据:

script/member_active/ods_load_log.sh

可以传参数确定日志,如果没有传参使用昨天日期

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#!/bin/bash

APP=ODS
hive=/opt/lagou/servers/hive-2.3.7/bin/hive

# 可以输入日期;如果未输入日期取昨天的时间
if [ -n "$1" ]
then
do_date=$1
else
do_date=`date -d "-1 day" +%F`
fi

# 定义要执行的SQL
sql=" alter table "$APP".ods_start_log add partition(dt='$do_date'); "

$hive -e "$sql"

json数据处理

数据文件中每行必须是一个完整的 json 串,一个 json串不能跨越多行。

Hive 处理json数据总体来说有三个办法:

  • 使用内建的函数get_json_object、json_tuple

  • 使用自定义的UDF

  • 第三方的SerDe

使用内建函数处理

  • get_json_object(string json_string, string path)

    返回值:String

    说明:解析json字符串json_string,返回path指定的内容;如果输入的json字符串无效,那么返回NUll;函数每次只能返回一个数据项;

  • json_tuple(jsonStr, k1, k2, …)

    返回值:所有的输入参数、输出参数都是String;

    说明:参数为一组键k1,k2,。。。。。和json字符串,返回值的元组。该方法比get_json_object高效,因此可以在一次调用中输入多个键;

explode,使用explod将Hive一行中复杂的 array 或 map 结构拆分成多行。

测试数据格式:

1
2
3
4
5
user1;18;male;{"id": 1,"ids": [101,102,103],"total_number": 3}
user2;20;female;{"id": 2,"ids": [201,202,203,204],"total_number": 4}
user3;23;male;{"id": 3,"ids": [301,302,303,304,305],"total_number": 5}
user4;17;male;{"id": 4,"ids": [401,402,403,304],"total_number": 5}
user5;35;female;{"id": 5,"ids": [501,502,503],"total_number": 3}

建表加载数据:

1
2
3
4
5
6
7
8
CREATE TABLE IF NOT EXISTS jsont1(
username string,
age int,
sex string,
json string
)row format delimited fields terminated by ';';

load data local inpath '/data/lagoudw/data/weibo.json' overwrite into table jsont1;

json的处理:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
-- get 单层值
select username, age, sex,
get_json_object(json, "$.id") id,
get_json_object(json, "$.ids") ids,
get_json_object(json, "$.total_number") num
from jsont1;

-- get 数组
select username, age, sex,
get_json_object(json, "$.id") id,
get_json_object(json, "$.ids[0]") ids0,
get_json_object(json, "$.ids[1]") ids1,
get_json_object(json, "$.ids[2]") ids2,
get_json_object(json, "$.ids[3]") ids3,
get_json_object(json, "$.total_number") num
from jsont1;

-- 使用 json_tuple 一次处理多个字段
select json_tuple(json, 'id', 'ids', 'total_number')
from jsont1;

-- 有语法错误
select username, age, sex,
json_tuple(json, 'id', 'ids', 'total_number')
from jsont1;

-- 使用 explode + lateral view
-- 在上一步的基础上,再将数据展开
-- 第一步,将 [101,102,103] 中的 [ ] 替换掉
-- select "[101,102,103]"
-- select "101,102,103"
select regexp_replace("[101,102,103]", "\\[|\\]", "");

-- 第二步,将上一步的字符串变为数组
select split(regexp_replace("[101,102,103]", "\\[|\\]", ""), ",");

-- 第三步,使用explode + lateral view 将数据展开
select username, age, sex, id, ids, num
from jsont1
lateral view json_tuple(json, 'id', 'ids', 'total_number') t1 as id, ids, num;

with tmp as(
select username, age, sex, id, ids, num
from jsont1
lateral view json_tuple(json, 'id', 'ids', 'total_number') t1 as id, ids, num
)
select username, age, sex, id, ids1, num
from tmp
lateral view explode(split(regexp_replace(ids, "\\[|\\]", ""), ",")) t1 as ids1;

小结:json_tuple 优点是一次可以解析多个json字段,对嵌套结果的解析操作复杂;

使用UDF处理

自定义UDF处理json串中的数组。自定义UDF函数:

  • 输入:json串、数组的key

  • 输出:字符串数组

pom文件增加依赖

1
2
3
4
5
6
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>2.3.7</version>
<scope>provided</scope>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package cn.lagou.dw.hive.udf;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Strings;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.junit.Test;
import java.util.ArrayList;

public class ParseJsonArray extends UDF {
public ArrayList<String> evaluate(String jsonStr, String arrKey){
if (Strings.isNullOrEmpty(jsonStr)) {
return null;
}
try{
JSONObject object = JSON.parseObject(jsonStr);
JSONArray jsonArray = object.getJSONArray(arrKey);
ArrayList<String> result = new ArrayList<>();
for (Object o: jsonArray){
result.add(o.toString());
}
return result;
} catch (JSONException e){
return null;
}
}

@Test
public void JunitParseJsonArray(){
String str = "{\"id\": 1,\"ids\": [101,102,103],\"total_number\": 3}";
String key = "ids";
ArrayList<String> evaluate = evaluate(str, key);
System.out.println(JSON.toJSONString(evaluate));
}
}

使用自定义 UDF 函数:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
-- 添加开发的jar包(在Hive命令行中)
add jar /data/lagoudw/jars/cn.lagou.dw-1.0-SNAPSHOT-jar-with- dependencies.jar;

-- 创建临时函数。指定类名一定要完整的路径,即包名加类名
create temporary function lagou_json_array as "cn.lagou.dw.hive.udf.ParseJsonArray";

-- 执行查询
-- 解析json串中的数组
select username, age, sex, lagou_json_array(json, "ids") ids
from jsont1;

-- 解析json串中的数组,并展开
select username, age, sex, ids1
from jsont1
lateral view explode(lagou_json_array(json, "ids")) t1 as ids1;

-- 解析json串中的id、num
select username, age, sex, id, num
from jsont1
lateral view json_tuple(json, 'id', 'total_number') t1 as id, num;

-- 解析json串中的数组,并展开
select username, age, sex, ids1, id, num
from jsont1
lateral view explode(lagou_json_array(json, "ids")) t1 as ids1
lateral view json_tuple(json, 'id', 'total_number') t1 as id, num;

使用SerDe处理

序列化是对象转换为字节序列的过程;反序列化是字节序列恢复为对象的过程;

对象的序列化主要有两种用途:

  • 对象的持久化,即把对象转换成字节序列后保存到文件中

  • 对象数据的网络传送

SerDe 是Serializer 和 Deserializer 的简写形式。Hive使用Serde进行行对象的序列与反序列化。最后实现把文件内容映射到 hive 表中的字段数据类型。SerDe包括Serialize/Deserilize 两个功能:

  • Serialize把Hive使用的java object转换成能写入HDFS字节序列,或者其他系统能识别的流文件

  • Deserilize把字符串或者二进制流转换成Hive能识别的java object对象

Read : HDFS files => InputFileFormat => <key, value> => Deserializer => Row object

Write : Row object => Seriallizer => <key, value> => OutputFileFormat => HDFS files

常见:https://cwiki.apache.org/confluence/display/Hive/DeveloperGuide#DeveloperGuide-HiveSerDe

Hive本身自带了几个内置的SerDe,还有其他一些第三方的SerDe可供选择。

1
2
3
4
5
6
7
8
9
10
11
12
create table t11(
id string
)
stored as parquet;

create table t12(
id string
)
stored as ORC;

desc formatted t11;
desc formatted t12;

LazySimpleSerDe(默认的SerDe)

ParquetHiveSerDe

OrcSerde

对于纯 json 格式的数据,可以使用 JsonSerDe 来处理。

1
2
3
4
5
{"id": 1,"ids": [101,102,103],"total_number": 3}
{"id": 2,"ids": [201,202,203,204],"total_number": 4}
{"id": 3,"ids": [301,302,303,304,305],"total_number": 5}
{"id": 4,"ids": [401,402,403,304],"total_number": 5}
{"id": 5,"ids": [501,502,503],"total_number": 3}
1
2
3
4
5
6
7
8
create table jsont2(
id int,
ids array<string>,
total_number int
)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe';

load data local inpath '/data/lagoudw/data/json2.dat' into table jsont2;

各种Json格式处理方法小结:

1、简单格式的json数据,使用get_json_object、json_tuple处理

2、对于嵌套数据类型,可以使用UDF

3、纯json串可使用JsonSerDe处理更简单