大数据离线业务场景中的增量技术
Posted 杀智勇双全杀
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了大数据离线业务场景中的增量技术相关的知识,希望对你有一定的参考价值。
大数据离线业务场景中的增量技术
业务需求
离线中的业务需要一般需要离线采集、离线计算、离线结果保存。实时可以替代离线,但是离线不能替代实时。
离线
以时间点为触发条件(以时间为单位)来实现数据的处理(采集、计算等)。时效性比较低,一般不会高于分钟级。Hadoop生态圈从Sqoop、HDFS、Hive、MapReduce、SparkTez、Impala等都是离线工具。处理的周期一般是T+1(按天处理),也可以每小时、每月、每年处理一次。
实时
以数据变化为触发条件(以数据为单位)实现数据的处理(采集、计算等)。时效性非常高,一般更新周期为ms级(但是不可能达到工业上的us级),已经可以满足大部分使用场景。实时生态圈的Flume、Canal、Kafka、SparkStreaming/Flink、Redis、Hbase等都是离线工具。产生一条数据就会立即处理一条数据。
增量
每次对最新(新增Insert、更新Update)的数据进行处理。
全量
每次都所有数据进行处理,一般是数据迁移(首次建仓时进行数据同步)、维度表更新时使用。
增量采集方案
Flume增量采集
Flume基础讲述过可以使用exec
和taildir
分别实时动态监听单个文件尾部(配合tail
命令,自动读取文件的尾部)、实时动态监听目录内多个文件(taildir_position.json
记录文件的采集位置)的变动。
Sqoop增量采集
Sqoop基础讲述过可以使用Sqoop实现离线数据的增量采集。
append(按照某一列自增的int值)
必须有一列自增的int值,必须有主键。只能采集新增(Insert)的数据。
lastmodifield(按照数据变化的时间列的值)
必须有一列时间列,时间列随着数据的更新而自动更新。能采集新增(Insert)和更新(Update)的数据
where过滤(指定目录分区采集到对应的HDFS目录)
表中需要有这2个字段:
create_time:创建时间,用于标记新增的数据。
update_time:更新时间,用于标记更新的数据。
例如:使用-e传递shell的变量:
-e "select * from table where substr(create_time,1,10) = '2021-05-15' or substr(update_time,1,10) = '2021-05-16'“
就可以处理昨天的数据。。。
注意事项
增量要求目录提前存在,以便追加新增的数据进入。如果没有使用官方提供的增量方式(也就是append和 lastmodifield的方式。。。where过滤是用户自己实现的),目录不能提前存在(del掉target—dir又会导致之前存储的数据丢失)。按时间(日期)构建分区存储场景:
--target-dir /nginx/log/2021-05-15/
每次采集的目录不同即可解决这个问题。
将数据采集到Hive的分区表
Sqoop官方方案
–hive-partition-key daystr:指定分区的字段是哪个字段。
–hive-partition-value 2021-05-15:指定导入哪个分区。
这种方式,Sqoop会根据指定的参数,在HDFS中创建一个key=value的目录:table/daystr=2021-05-15。用户只需要在Hive中加载分区即可。但是这种方式不能创建多级分区。
手动指定HDFS目录
--target-dir /nginx/log/daystr=2021-05-15/hourstr=00
增量采集实现
mysql中准备数据:
create database if not exists db_order;
use db_order;
drop table if exists tb_order;
create table tb_order(
id varchar(10) primary key,
pid varchar(10) not null,
userid varchar(10) not null,
price double not null,
create_time varchar(20) not null
);
insert into tb_order values('o00001','p00001','u00001',100,'2021-05-13 00:01:01');
insert into tb_order values('o00002','p00002','u00002',100,'2021-05-13 10:01:02');
insert into tb_order values('o00003','p00003','u00003',100,'2021-05-13 11:01:03');
insert into tb_order values('o00004','p00004','u00004',100,'2021-05-13 23:01:04');
insert into tb_order values('o00005','p00005','u00001',100,'2021-05-14 00:01:01');
insert into tb_order values('o00006','p00006','u00002',100,'2021-05-14 10:01:02');
insert into tb_order values('o00007','p00007','u00003',100,'2021-05-14 11:01:03');
insert into tb_order values('o00008','p00008','u00004',100,'2021-05-14 23:01:04');
Sqoop第一次采集:
sqoop import \\
--connect jdbc:mysql://node3:3306/db_order \\
--username root \\
--password-file file:///export/data/sqoop.passwd \\
--query "select * from tb_order where substring(create_time,1,10) = '2021-05-14' and \\$CONDITIONS " \\
--delete-target-dir \\
--target-dir /nginx/logs/tb_order/daystr=2021-05-14 \\
--fields-terminated-by '\\t' \\
-m 1
MySQL产生新数据:
insert into tb_order values('o00009','p00005','u00001',100,'2021-05-15 00:01:01');
insert into tb_order values('o00010','p00006','u00002',100,'2021-05-15 10:01:02');
insert into tb_order values('o00011','p00007','u00003',100,'2021-05-15 11:01:03');
insert into tb_order values('o00012','p00008','u00004',100,'2021-05-15 23:01:04');
Sqoop再次采集:
sqoop import \\
--connect jdbc:mysql://node3:3306/db_order \\
--username root \\
--password-file file:///export/data/sqoop.passwd \\
--query "select * from tb_order where substring(create_time,1,10) = '2021-05-15' and \\$CONDITIONS " \\
--delete-target-dir \\
--target-dir /nginx/logs/tb_order/daystr=2021-05-15 \\
--fields-terminated-by '\\t' \\
-m 1
即可实现增量采集。
增量处理
beeline执行:
use default;
drop table default.tb_order;
create table if not exists default.tb_order(
id string ,
pid string,
userid string,
price double ,
create_time string
)
partitioned by (daystr string)
row format delimited fields terminated by '\\t'
location '/user/hive/warehouse/tb_order';
在Hive中建表。假设需要处理之前的数据,由于ETL发生在进入Hive之前(当然ODS层→DW层也可能有ETL),ETL都是通过SpakrCore、MapReduce之类的程序处理的,这些程序一般会使用YARN托管运行:
yarn jar etl.jar main_class \\
#输入目录:数据采集的目录
/nginx/logs/tb_order/daystr=2021-05-14 \\
#输出目录:构建Hive表的目录
/user/hive/warehouse/tb_order/daystr=2021-05-14
日期当然也不会写死:
//Input
Path inputPath = new Path("/nginx/logs/tb_order/daystr="+yesterday)
TextInputFormat.setInputPaths(job,inputPath)
//Output
Path outputPath = new Path("/user/hive/warehouse/tb_order/daystr="+yesterday)
TextOutputFormat.setOutputPath(job,outputPath)
MapReduce中是这样指定路径的。
如果直接把数据放到HDFS目录:
hdfs dfs -cp /nginx/logs/tb_order/daystr=2021-05-14 /user/hive/warehouse/tb_order/
数据当然是不会被HDFS识别的(分区表不是Hive创建的,缺少元数据)。使用:
show partitions tb_order;
select * from tb_order;
都看不到相关内容。
手动修改元数据
使用:
msck repair table tb_order;
修改Hive元数据(添加分区)后可以看到所需内容。按照分区表的查询方式即可查询数据:
select
daystr,
count(id) as order_number,
sum(price) as order_price
from default.tb_order
where daystr='2021-05-14'
group by daystr;
添加分区元数据信息
假设ETL得到Hive数据:
hdfs dfs -cp /nginx/logs/tb_order/daystr=2021-05-15 /user/hive/warehouse/tb_order/
使用:
alter table tb_order add if not exists partition(daystr='2021-05-15');
如果不存在分区信息便添加分区元数据。
Hive中HDFS上目录的名称必须为分区字段=值(例如:/user/hive/warehouse/tb_order/daystr=2021-05-15
)如果出现不满足的情况(例如:/user/hive/warehouse/tb_order/2021-05-15
)就需要通过location关键字手动指定分区对应的HDFS:
alter table tb_order add if not exists partition(daystr='2021-05-15') location '/user/hive/warehouse/tb_order/2021-05-15';
之后便可以实现增量处理:
select
daystr,
count(id) as order_number,
sum(price) as order_price
from default.tb_order
where daystr='2021-05-15'
group by daystr;
增量导出
Hive中建立APP层结果表:
drop table if exists tb_order_rs;
create table if not exists default.tb_order_rs(
daystr string,
order_number int,
order_price double
)
row format delimited fields terminated by '\\t';
MySQL中建立导出结果表:
use db_order;
drop table if exists db_order.tb_order_rs;
create table db_order.tb_order_rs(
daystr varchar(20) primary key,
order_number int,
order_price double
);
将第一次分析的结果写入APP层结果表:
insert into table tb_order_rs
select
daystr,
count(id) as order_number,
sum(price) as order_price
from default.tb_order
where daystr='2021-05-14'
group by daystr;
第一次导出到MySQL:
sqoop export \\
--connect jdbc:mysql://node3:3306/db_order \\
--username root \\
--password 123456 \\
--table tb_order_rs \\
--hcatalog-database default \\
--hcatalog-table tb_order_rs \\
--input-fields-terminated-by '\\t' \\
--update-key daystr \\
--update-mode allowinsert \\
-m 1
第二次分析的结果写入APP层结果表:
insert into table tb_order_rs
select
daystr,
count(id) as order_number,
sum(price) as order_price
from default.tb_order
where daystr='2021-05-15'
group by daystr;
还是使用相同的代码便可增量导出到MySQL:
sqoop export \\
--connect jdbc:mysql://node3:3306/db_order \\
--username root \\
--password 123456 \\
--table tb_order_rs \\
--hcatalog-database default \\
--hcatalog-table tb_order_rs \\
--input-fields-terminated-by '\\t' \\
--update-key daystr \\
--update-mode allowinsert \\
-m 1
脚本
ETL工具一般自己就能实现定时执行任务(Kettle中叫”作业“),但并不是所有程序都可以自动执行。Shell脚本这篇讲述了如何写简单的Shell脚本。
增量采集脚本
mkdir /export/data/shell
vim /export/data/shell/01.collect.sh
插入内容:
#!/bin/bash
#step1:先获取要采集的数据时间。如果没有给参数,默认处理昨天的日期,否则处理参数对应的日期
if [ $# -ne 0 ]
then
#参数个数不为0
if [ $# -ne 1 ]
then
echo "参数至多只能有一个,为处理的日期,请重新运行!"
exit 100 #手动指定错误码标记出现故障的位置
else
#参数个数只有1个,就用第一个参数作为处理的日期
yesterday=$1
fi
else
#参数个数为0,默认处理昨天的日期
yesterday=`date -d '-1 day' +%Y-%m-%d`
fi
echo "step1:要处理的日期是:${yesterday}"
echo "step2:开始运行采集的程序"
#step2:运行增量采集
SQOOP_HOME=/export/server/sqoop-1.4.6-cdh5.14.0
$SQOOP_HOME/bin/sqoop import \\
--connect jdbc:mysql://node3:3306/db_order \\
--username root \\
--password-file file:///export/data/sqoop.passwd \\
--query "select * from tb_order where substring(create_time,1,10) = '${yesterday}' and \\$CONDITIONS " \\
--delete-target-dir \\
--target-dir /nginx/logs/tb_order/daystr=${yesterday} \\
--fields-terminated-by '\\t' \\
-m 1
echo "step2:采集的程序运行结束"
echo "step3:开始运行ETL"
#模拟ETL的过程,将采集的新增的数据移动到表的目录下
HADOOP_HOME=/export/server/hadoop-2.6.0-cdh5.14.0
#先判断结果是否存在,如果已经存在,先删除再移动。-e是判断是否存在
$HADOOP_HOME/bin/hdfs dfs -test -e /user/hive/warehouse/tb_order/daystr=${yesterday}
if [ $? -eq 0 ]
then
#存在
$HADOOP_HOME/bin/hdfs dfs -rm -r /user/hive/warehouse/tb_order/daystr=${yesterday}
$HADOOP_HOME/bin/hdfs dfs -cp /nginx/logs/tb_order/daystr=${yesterday} /user/hive/warehouse/tb_order/
else
#不存在
$HADOOP_HOME/bin/hdfs dfs -cp /nginx/logs/tb_order/daystr=${yesterday} /user/hive/warehouse/tb_order/
fi
echo "step3:ETL结束"
增量处理脚本
vim /export/data/shell/02.analysis.sh
vim /export/data/shell/02.analysis.sql
分别在2个文件中写入:
#!/bin/bash
#step1:先获取要采集的数据时间。如果没有给参数,默认处理昨天的日期。否则,处理参数对应的日期
if [ $# -ne 0 ]
then
#参数个数不为0
if [ $# -ne 1 ]
then
echo "参数至多只能有一个,为处理的日期,请重新运行!"
exit 100
else
#参数个数只有1个,用第一个参数作为处理的日期
yesterday=$1
fi
else
#参数个数为0,默认处理昨天的日期
yesterday=`date -d '-1 day' +%Y-%m-%d`
fi
echo "step1:要处理的日期是:${yesterday}"
echo "step2:开始运行分析"
#step2:运行分析程序
HIVE_HOME=/export/server/hive-1.1.0-cdh5.14.0
$HIVE_HOME/bin/hive --hiveconf yest=${yesterday} -f /export/data/shell/02.analysis.sql
echo "step2:分析的程序运行结束"
和:
create table if not exists default.tb_order(
id string ,
pid string,
userid string,
price double ,
create_time string
)
partitioned by (daystr string)
row format delimited fields terminated by '\\t'
location '/user/hive/warehouse/tb_order';
alter table default.tb_order add if not exists partition (daystr='${hiveconf:yest}');
create table if not exists default.tb_order_rs(
daystr string,
order_number int,
order_price double
)
row format delimited fields terminated by '\\t';
insert into table default.tb_order_rs
select
daystr,
count(id) as order_number,
sum(price) as order_price
from default.tb_order
where daystr='${hiveconf:yest}'
group by daystr;
即可实现shell脚本调用文件SQL。
增量导出脚本
vim /export/data/shell/03.export.sh
写入:
#!/bin/bash
echo "step1:开始运行导出的程序"
#step2:运行增量采集
SQOOP_HOME=/export/server/sqoop-1.4.6-cdh5.14.0
$SQOOP_HOME/bin/sqoop export \\
--connect jdbc:mysql://node3:3306/db_order \\
--username root \\
--password 123456 \\
--table tb_order_rs \\
--hcatalog-database default \\
--hcatalog-table tb_order_rs \\
--input-fields-terminated-by '\\t' \\
--update-key daystr \\
--update-mode allowinsert \\
-m 1
echo "step1:导出的程序运行结束"
即可将数据增量导出到MySQL。
尾言
虽然写了Shell脚本和SQL文件,但是现在只能实现手动调度进行增量采集、增量处理、增量导出,依然不能自动化运行。
基于Hue和Oozie实现自动化调度继续讲述。
以上是关于大数据离线业务场景中的增量技术的主要内容,如果未能解决你的问题,请参考以下文章
尚大视频开放周基于阿里云搭建数据仓库(离线)项目视频开放下载!
大数据技术之_18_大数据离线平台_01_项目开发流程+大数据常用应用/分析平台/业务处理方式+数据分析平台的数据来源+数据处理的流程+项目集群的规模+需求分析+技术架构/选型