自行车销售项目
本文是对Adventure Works案例的一个总结，记录了整个项目需求分析与实现的过程，主要任务是使用Hive SQL完成ETL过程，实现hive数据仓库，并连接到Tableau实现可视化。中间加入了个人的一些思考，最终将整个分析成果展示出来。
项目背景及目标：
Adventure Works Cycle是国内一家生产和销售自行车及和相关配件的制造公司。前期主要是分销商模式，2019年开始通过公司自有网站获取线上客户进一步扩大市场，根据公司的销售数据，制作可视化看板，帮助运营人员自主分析和持续监测产品销售情况，及时为公司业务优化提供辅助决策，实现高效经营。
项目分析实现过程：
1、初步了解数据：
先查看项目数据库里的数据表，找到相关的数据，初步了解数据，有四个基础数据表，分别为dim_date_df 日期表，ods_customer 每日新增客户信息表，ods_sales_order 订单明细表，在Powerbi中建模观察数据之间的关系。
2、根据目标分析指标：
要制作看板，根据看板遵循金字塔结构，不同角色关注的指标不同，高层关注的是结果，所以要有一个体现结果性的KPI指标分析；中高层关注过程，所以要体现运营过程的分析，要有监控看板，包括跟踪、对比等；基层关注的是业务运营指标的绩效分析
从三方面入手：
a、制作一个区域细分看板，按照日期，区域、城市、产品类别划分，展示其对应的产品订单量、客户数、订单金额等情况
b、制作一个数据监控看板，展示订单量，订单金额随时间的变化，以平均线为基准进行数据监控
c、制作一个周期性看板，展示每日或月的总体情况（订单金额和订单量、平均订单价）以及当日同比结果，不同维度订单量占比分布（按照产品类别省份），不同维度销售排名（产品类别，产品子类）
需要的数据指标：
每日的销售金额，客户数量，客单价、每日同比（按时间）、按日期/区域/城市/产品逐级类别分类后的订单量、客户数、订单金额
3、用Python处理数据获取指标
每日销售环比表dw_order_by_day：根据create_date分组求得每日的总销售金额和客户数量，客单价，再和日期维度dim_date_df表通过create_date融合后，利用Python的pct_change()函数求得每日环比再追加至表中
具体如下：
#导入相关模块
import pandas as pd
import pymysql
import random
from sqlalchemy import create_engine
pymysql.install_as_MySQLdb()
import datetime
import warnings
warnings.filterwarnings("ignore")
# 连接adventure_ods库
adventure_ods = create_engine('mysql://*****主库，只读)
# 连接datafrog05_adventure_new库
adventure_dw = create_engine('mysql://****部门库，可增删改查)
"""step1：读取ods_sales_orders(订单明细表)，生成sum_amount_order(销量订单聚合表)，求总销售金额/客单价"""
ods_sales_orders = pd.read_sql_query("select * from ods_sales_orders ", con=adventure_ods)
# 根据create_date分组求总销售金额及客户数量
sum_amount_customer = ods_sales_orders.groupby('create_date').agg({'unit_price': sum, 'customer_key': pd.Series.nunique}).reset_index()
# 修改对应列名
sum_amount_customer.rename(columns={'unit_price': 'sum_amount', 'customer_key': 'sum_order'}, inplace=True)
# 新增amount_div_order客单价列
sum_amount_customer['amount_div_order'] = sum_amount_customer['sum_amount'] / sum_amount_customer['sum_order']
sum_amount_order = sum_amount_customer
"""step2:读取dim_date_df日期维度表"""
dim_date_df = """select create_date,
is_current_year,
is_last_year,
is_yesterday,
is_today,
is_current_month,
is_current_quarter
from dim_date_df"""
dim_date_df = pd.read_sql_query(dim_date_df, con=adventure_ods)
"""step3：进行数据的融合，生成dw_order_by_day表"""
# 通过主键create_date连接日期维度
dw_order_by_day = pd.merge(sum_amount_order, dim_date_df, on='create_date', how='inner')
"""step4：删除旧的dw_order_by_day(每日环比表)"""
# 因为删除插入更新操作没有返回值，程序会抛出ResourceClosedError，并终止程序。使用try捕捉此异常。
try:
pd.read_sql_query("Truncate table dw_order_by_day_{}".format(STUDENT_NAME), con=adventure_dw)
except Exception as e:
print("删除旧的dw_order_by_day表，error：{}".format(e)
"""step4：存储新的dw_order_by_day """
#增添环比值amount_diff
# pct_change()表示当前元素与先前元素的相差百分比，默认竖向，例：前面元素x，当前元素y，公式 result = (y-x)/x
dw_order_by_day['amount_diff'] = dw_order_by_day['sum_amount'].pct_change().fillna(0)
# 追加数据至dw_order_by_day
dw_order_by_day.to_sql('dw_order_by_day_{}'.format(STUDENT_NAME), con=adventure_dw,
if_exists='replace', index=False)
dw_order_by_day.head()
今日销售环比表dw_amount_diff：根据dw_order_by_day生成dw_amount_diff当日维度表（按当天、昨天、当月、当季、当年的同比），求取各阶段的总金额、订单数的同期对比数据,结果如下
"""step1:求取各阶段的总金额"""（以当天为例，其余类同）
"""当天"""
# 当天的总金额
today_amount = dw_order_by_day[dw_order_by_day['is_today'] == 1]['sum_amount'].sum()
# 去年同期的日期维度
before_year_today = list(dw_order_by_day[dw_order_by_day['is_today'] == 1]
['create_date'] + datetime.timedelta(days=-365))
# 去年同期总金额
before_year_today_amount = dw_order_by_day[dw_order_by_day['create_date'].isin(before_year_today)]['sum_amount'].sum()
"""step2:求取各阶段的总订单数"""
"""当天"""
# 当天的总订单数
today_order = dw_order_by_day[dw_order_by_day['is_today'] == 1]['sum_order'].sum()
# 去年同期总订单数
before_year_today_order = dw_order_by_day[dw_order_by_day['create_date'].isin(
before_year_today)]['sum_order'].sum()
"""step3:求取各阶段的总金额、订单数的同期对比数据""" 现在值 - 过去值 / 过去值 = 现在值 / 过去值 - 1
# 做符号简称，横向提取数据方便
amount_dict = {'today_diff': [today_amount / before_year_today_amount - 1,
today_order / before_year_today_order - 1,
(today_amount / today_order) / (before_year_today_amount /
before_year_today_order) - 1]
今日销售聚合表dw_customer_order：按照 订单日期/产品名/产品子类/产品类别/所在区域/所在省份/所在城市的逐级聚合，合并日期维度表，获得订单总量/客户总量/销售总金额+当日日期维度，结果如下：
"""step1:读取最新日期的ods_sales_orders（订单明细表) """
ods_sales_orders = \
"""select sales_order_key,
create_date,
customer_key,
english_product_name,
cpzl_zw,
cplb_zw,
unit_price
from ods_sales_orders where create_date =(
select create_date
from dim_date_df
order by create_date desc
limit 1) """
ods_sales_orders = pd.read_sql_query(ods_sales_orders, con=adventure_ods)
"""step2：读取每日新增用户表ods_customer"""
ods_customer = """
select customer_key,
chinese_territory,
chinese_province,
chinese_city
from ods_customer"""
ods_customer = pd.read_sql_query(ods_customer, con=adventure_ods)
"""step3:读取日期维度表dim_date_df"""
dim_date_df = """
select create_date,
is_current_year,
is_last_year,
is_yesterday,
is_today,
is_current_month,
is_current_quarter
from dim_date_df"""
dim_date_df = pd.read_sql_query(dim_date_df, con=adventure_ods)
"""step4:进行数据的聚合"""
# 通过客户id连接表
sales_customer_order = pd.merge(ods_sales_orders, ods_customer, left_on='customer_key',right_on='customer_key', how='left')
# 提取订单主键/订单日期/客户编号/产品名/产品子类/产品类别/产品单价/所在区域/所在省份/所在城市
sales_customer_order = sales_customer_order[["sales_order_key", "create_date", "customer_key","english_product_name", "cpzl_zw", "cplb_zw", "unit_price",
"chinese_territory","chinese_province","chinese_city"]]
# 形成按照订单日期/产品名/产品子类/产品类别/所在区域/所在省份/所在城市的逐级聚合表，获得订单总量/客户总量/销售总金额
sum_customer_order = sales_customer_order.groupby(["create_date", "english_product_name", "cpzl_zw", "cplb_zw","chinese_territory", "chinese_province",
"chinese_city"], as_index=False).agg({'sales_order_key': pd.Series.nunique, 'customer_key': pd.Series.nunique,"unit_price": "sum"}).rename(columns={'sales_order_key': 'order_num','customer_key': 'customer_num', 'unit_price': 'sum_amount',"english_product_name": "product_name"})
# 转化订单日期为字符型格式
sum_customer_order['create_date'] = sum_customer_order['create_date'].apply(lambda x: x.strftime('%Y-%m-%d'))
# 获取当日日期维度
dw_customer_order = pd.merge(sum_customer_order, dim_date_df, on='create_date', how='inner')
dw_customer_order.head()
- 在数据量增大到千万级别时，在用Python连接MySQL中进行查数和聚合处理等操作会特别缓慢，需要将数据放在hive数据库下进行处理，具体实现过程如下
1 、数据处理（hive数据仓库的实现）
处理过程分为三个步骤：1、使用sqoop从Mysql数据库导数到（hdfs系统）的hive数据库生成ods层 2、使用hive库根据业务需求进行数据汇总加工生成dw层 3、把加工后的数据使用sqoop推送到MySQL库
具体操作如下：
**步骤 一 同步Mysql数据库ods层数据到hadoop生态体系下的的hive数据库
MySQL数据库的ods库中有三张基础数据表：ods_sales_orders、ods_customer、dim_date_df表，先在文本编辑器上编写三个shell脚本，打开Linux窗口，运行shell脚本，实现sqoop导数，可以同步MySQL数据库的ods层数据到hadoop生态体系下的hdfs文件体系系统中的hive数据库形成hive数据库的ods层
sqoop_dim_date_df.sh、sqoop_ods_customer.sh、sqoop_ods_sales_oreder.sh 代码如下（举例其中一个）
#### 导入查询出来的数据到Hive 运行时#后面内容会去除
## 导入dim_date_df表 （sqoop_dim_date_df.sh）
#!/bin/sh
hive -e "drop table if exists ods.dim_date_df" # 删除hive原有的旧表
sqoop import \
--hive-import \
## 告诉jdbc，连接mysql的url
--connect jdbc:mysql://***ip*****/adventure_ods?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&dontTrackOpenResources=true&defaultFetchSize=50000&useCursorFetch=true" \
--driver com.mysql.jdbc.Driver \ # Hadoop根目录
--username ***** \ ## 连接mysql的用户名
--password ******* \ ## 连接mysql的密码
--query \ "select * from dim_date_df where "'$CONDITIONS'" " \
--fetch-size 50000 \ ## 一次从数据库读取 n 个实例，即n条数据
--hive-table ods.dim_date_df \ ## 创建dim_date_df表（默认也会自动创建表）
--hive-drop-import-delims \ ## 在导入数据到hive时，去掉数据中的\r\n\013\010这样的字符
--delete-target-dir \ ## 如果目标文件已存在就把它删除
--target-dir /user/hadoop/sqoop/dim_date_df \ ## 指定的目录下面并没有数据文件，数据文件被放在了hive默认/user/hadoop/sqoop/dim_date_df下
-m 1 ## 迁移过程使用1个map（开启一个线程）
步骤二 使用hive数据库加工ods层数据生成dw 层数据
同上分析的指标进行加工，下面是shell脚本的代码是做聚合的HIVESQL语句，（举例其一）
#!/bin/sh
hive -v -e "
drop table if exists ods.dw_order_by_day;
CREATE TABLE ods.dw_order_by_day(## hive创建表的语句
create_date string,
is_current_year bigint,
is_last_year bigint,
is_yesterday bigint,
is_today bigint,
is_current_month bigint,
is_current_quarter bigint,
sum_amount double,
sum_order bigint);
## 这里是hive的查询语句，因为做聚合需要关联多张表做聚合，这里使用with查询来提高查询性能
with dim_date as
(select create_date,
is_current_year,
is_last_year,
is_yesterday,
is_today,
is_current_month,
is_current_quarter
from ods.dim_date_df),
sum_day as
(select create_date,
sum(unit_price) as sum_amount,
count(customer_key) as sum_order
from ods.ods_sales_orders
group by create_date)
## 查询的数据插入表中
insert into ods.dw_order_by_day
select b.create_date,
b.is_current_year,
b.is_last_year,
b.is_yesterday,
b.is_today,
b.is_current_month,
b.is_current_quarter,
a.sum_amount,
a.sum_order
from sum_day as a
inner join dim_date as b
on a.create_date=b.create_date
"
- 以上的命令中后面的##部分是注释，执行的时候需要删掉
步骤三 使用sqoop导出数据到MySQL数据库
在导出数据到mysql前，先在mysql中已经创建好三个表dw_customer_order、dw_order_by_day、dw_amount_diff
导出过程如下(举例其一)：
导出数据到 mysql 表中
sqoop export --connect "jdbc:mysql://**ip**/库名" \
--username ** \
--password ** \
--table dw_customer_order \
--export-dir /user/hive/warehouse/ods.db/dw_customer_order \
--input-null-string "\\\\N" \
--input-null-non-string "\\\\N" \
--input-fields-terminated-by "\001" \
--input-lines-terminated-by "\\n" \
-m 1
2、连接MySQL数据库进行可视化报表制作
1、周期性看板（日报）的设计过程及结果展示 设计流程：
结果展示：
2、数据监控看板（日报）的设计过程及结果展示
设计流程：
结果展示：
3、区域细分看板（日报）的设计过程及结果展示 设计流程：
结果展示：
整体结果显示：
知识点
1 、linux系统和hive数据库和Hadoop生态体系下的hdfs系统关系
- 总结：hive是基于Hadoop生态体系存在的数据库，是大数据组件
2、维度建模：围绕一个事实表分拆不同维度建立关系
需要两类表，维度表和事实表
维度表：分析主题所属类型的描述，
事实表：分析主题的度量，一般不断增长，
有三种模式，雪花模式，星形模式，星座模式
3、tableau
a、tableau中的排序前N,先排序后，需要创建排名字段，设置index（）函数，再在筛选器中选择1~N即可
但如果是多维度的排序，需要再行中增加排名度量才能同时进行排序各种度量，如客户量、订单量、订单金额等
b、时间标题设置时，一般需先改为日期为日期格式，max函数选出当前更新日期
