• MySQL → HDFS：需要使用 Sqoop 命令。
• HDFS → 数据仓库 ODS 层：需要使用 Hive 的 Alter 命令。
• 数据仓库 ODS 层 → 数据仓库 DWD 层：需要使用 Hive SQL。
• 数据仓库 DWD 层 → 数据仓库 APP 层：需要使用 Hive SQL。

create database ods_mall;
create database dwd_mall;
create database app_mall;

（5）然后创建各个层级数据库所用到的 hive 表： 

• ods_user_order 表存放从 MySQL 到 HDFS 的原始订单数据：

create external table ods_mall.ods_user_order(
    order_id int,
    order_date string,
    user_id int,
    order_money decimal(10, 2),
    order_type string,
    order_status string,
    pay_id int,
    update_time timestamp
)
partitioned by(dt string)
row format delimited
fields terminated by '\t'
location '/data/ods/user_order';

• dwd_user_order 表存放的是 ods_user_order 经过清洗后的订单数据：

create external table dwd_mall.dwd_user_order(
    order_id int,
    order_date string,
    user_id int,
    order_money decimal(10, 2),
    order_type string,
    order_status string,
    pay_id int,
    update_time timestamp
)
partitioned by(dt string)
row format delimited
fields terminated by '\t'
location '/data/dwd/user_order';

• app_gmv 表存放的是通过对 dwd_user_order 表数据进行汇总计算得到的最终总成交额数据：

create external table app_mall.app_gmv (
  gmv decimal(10, 2)
)
partitioned by(dt string)
location '/data/app/app_gmv';

3，创建 collect.job 文件

# collect.job
# 将MySQL中的数据导入HDFS
type=command
command=sh /data/soft/warehouse_job/collect_mysql.sh

（2）在 collect.job 中需要使用到 collect_mysql.sh 这个脚本。这个脚本里面是 Sqoop 的采集数据命令，脚本内容如下：

# 每天凌展执行一次
# 默认获取昨天的日期，也支持传参指定一个自期
if [ "z$1" = "z" ]
then
  dt=`date +%Y%m%d --date="1 days ago"`
else
  dt=$1
fi
 
# 转换日期格式，将 20240201 转换为 2024-02-01
dt_new=`date +%Y-%m-%d --date="${dt}"`
 
# Hive SQL语句
user_order_sql="select order_id,order_date,user_id,order_money,order_type,order_status,pay_id,update_time from user_order where order_date > '${dt_new} 00:00:00' and order_date <= '${dt_new} 23:59:59'"
 
# 路径前缀
path_prefix="hdfs://node1:9000/data/ods"
 
# 输出路径
user_order_path="${path_prefix}/user_order/${dt}"
 
# 采集数据
echo ”开始采集...”
echo "采集表：user_order"
sh /data/soft/warehouse_job/sqoop_collect_data_util.sh "${user_order_sql}" "${user_order_path}"
echo "结束采集..."

（3）在 collect_mysql.sh 脚本中，还需要用到 sqoop_collect_data_util.sh 脚本，脚本内容如下：

#!/bin/bash
# 将MySQL中的数据导入HDFS
if [ $# !=2 ]
then
  echo "参数异常：sqoop_collect_data_util.sh <sql><hdfs_path>"
  exit 100
fi
 
# 查询指定日期范围订单的SQL语句
# 例如：select id,name from user where id >1
sql=$1
 
# 导入HDFS的路径
hdfs_path=$2
 
sqoop import \
--connect jdbc:mysql://192.168.60.1:3306/hangge?serverTimezone=Asia/Shanghai \
--username root \
--password hangge1234 \
--target-dir "${hdfs_path}" \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by '\t' \
--query "${sql}"' and $CONDITIONS' \
--null-string '\\N' \
--null-non-string '\\N'

4，创建 ods.job 文件

# ods.job
# 关联ODS层的数据
type=command
dependencies=collect
command=sh /data/soft/warehouse_job/ods_mall_add_partition.sh

（2）在 ods.job 中，需要使用到 ods_mall_add_partition.sh 这个脚本。这个脚本中是向 Hive 外部表中添加分区的命令，脚本内容如下：

#!/bin/bash
# 给ODS层的表添加分区，这个脚本每天执行一次
# 每天凌晨添加昨天的分区，添加完分区之后再执行后面的计算脚本

# 默认获取昨天的日期，也支持传参指定一个日期
if [ "z$1" = "z" ]
then
  dt=`date +%Y%m%d --date="1 days ago"`
else
  dt=$1
fi

sh /data/soft/warehouse_job/add_partition.sh ods_mall.ods_user_order ${dt} ${dt}

（3）在 ods_mall_add_partition.sh 脚本中还需要用到 add_partition.sh 脚本。add_partition.sh 脚本的内容如下：

#!/bin/bash
# 给外部分区表添加分区
# 接收3个参数
#1: 表名
#2: 分区字段dt的值：格式20240101
#3: 分区路径(相对路径或者绝对路径都可以)

if [ $# != 3 ]
then
  echo "参数异常：add_partition,sh <tabkle_name><dt><path>"
  exit 100
fi

table_name=$1
dt=$2
path=$3

hive -e "
alter table ${table_name} add if not exists partition(dt='${dt}') location '${path}';
"

5，创建 dwd.job 文件

（1）在 dwd.job 文件中使用 Hive SQL 将数据仓库 ODS 层中清洗之后（去除 order_id 为空的数据）的结果数据添加到 DWD 层的 Hive 外部表中。文件内容如下：

# dwd.job
# 生成DWD层的数据
type=command
dependencies=ods
command=sh /data/soft/warehouse_job/dwd_mall_add_partition.sh

（2）在 dwd.job 文件中需要使用到 dwd_mall_add_partition.sh 这个脚本。这个脚本中是向 Hive 外部表的指定分区中添加数据的命令。脚本内容如下：

#!/bin/bash
# 基于ODS层的表进行清洗，将清洗之后的数据添加到DWD层对应表的对应分区中
#每天凌展执行一次

# 默认获取昨天的日期，也支持传参指定一个日期
if [ "z$1" = "z" ]
then
  dt=`date +%Y%m%d --date="1 days ago"`
else
  dt=$1
fi

hive -e "
insert overwrite table dwd_mall.dwd_user_order partition(dt='${dt}') select
  order_id,
  order_date,
  user_id,
  order_money,
  order_type,
  order_status,
  pay_id,
  update_time
from ods_mall.ods_user_order
where dt = '${dt}' and order_id is not null;
"

6，创建 app.job 文件

# app.job
# 生成APP层的数据
type=command
dependencies=dwd
command=sh /data/soft/warehouse_job/app_mall_add_partition.sh

（2）在 app.job 中，需要使用到 app_mall_add_partition.sh 这个脚本。这个脚本中是向 Hive 外部表的指定分区中添加数据的命令。脚本内容如下：

#!/bin/bash
# 需求：电商GMV
# 每天凌晨执行一次

# 默认获取昨天的日期，也支持传参指定一个日期
if [ "z$1" = "z" ]
then
  dt=`date +%Y%m%d --date="1 days ago"`
else
  dt=$1
fi

hive -e "
insert overwrite table app_mall.app_gmv partition(dt='${dt}') select
sum(order_money) as gmv
from dwd_mall.dwd_user_order
where dt = '${dt}';
"

7，创建项目并上传任务

（1）我们首先在 Azkaban 中创建一个名为 gmv_calc 的项目：

（2）然后我们将前面生成的 4 个 Job 文件打包成一个压缩包 gmv_calc.zip，并上传这个压缩包。

8，执行测试