BI-电商销售数据分析

时间：2023-03-16 00:34:02 | 来源：电子商务

时间：2023-03-16 00:34:02 来源：电子商务

目录

0 项目背景

1 数据需求分析

2 数据加工

3 ETL效率提升

4 自动化部署

5 Dashboard设计

---

0 项目背景

Adventure Work Cycle是一家大型跨国制造公司，该公司生产和销售金属、复合材料自行车。在全国各个市场，销售方式主要有两种，分销商模式和电商模式。前期主要是分销商模式，2018年公司实现财政收入目标后，2019年就开始通过公司自有网站获取线上商户进一步扩大市场。随着销售数据量日益增长，需要监控业务发展过程中的异常问题；为了能够一路了然业务情况，应该建立BI系统，形成固化日常看板，供运营和管理者决策。

1 数据需求分析

首先，一定要对数据需求进行分析；数据需要方的意图？需要哪些数据内容？能够拆解成哪些指标？要加工哪些数据？把这些问题理清楚，在搭建BI系统的时候才能有清晰的逻辑，“谋定而后动”说的就是这个道理。

公司的人事架构呈现金字塔结构，那么便可以借助这个金字塔对数据需求进行分析。不同的角色关注的指标不同；公司高层关注的是结果性指标，例如交易金额、订单量、订单均价等；公司中高层关注的是过程性指标，例如交易金额趋势、订单量趋势；公司基层关注的是运营性指标，例如各产品订单量排名、区域细分订单表。

通过数据需求分析可知，Dashboard需要展示3部分内容：周期性看板、数据监控看板、区域细分看板。对看板展示内容进行拆分，包括监控指标、筛选器、标题、色彩等，具体如下：

2 数据加工

明确数据需求后，便可对数据进行加工，提取需要的监控指标。在数据加工前，务必要熟悉数据，否则无从下手；熟悉数据最快的方式就是查看源数据层（ods层）的数据字典，理清数据表之间的关系。该项目的数据字典如下：

ods_sales_orders是订单明细表，ods_customer是每日新增用户表，dim_date_df是日期维度表，三张表存在连接关系，通过表连接、聚合等操作可以得到想要的监控指标。ods_sales_orders和ods_customer通过customer_key连接，得到的表再通过create_date和dim_date_df连接，可得到dw_customer_order（时间_地区_产品聚合表）；ods_sales_orders和dim_date_df通过create_date连接，得到dw_order_by_day（每日环比表）；dw_order_by_day进一步加工便可得到dw_amount_diff（当日维度表，即同比表）。

有了思路后，就可以进行数据加工了，将源数据层（ods层）的数据加工至数据仓库（dw层），以便于后续开展数据分析工作。以加工dw_customer_order表的代码为例：

# 导入所需要的库import numpy as npimport pandas as pdimport pymysqlpymysql.install_as_MySQLdb()from sqlalchemy import create_engineimport datetime# 连接数据库（源数据层和数据仓库）AdventureOds = create_engine("mysql://UserName_1:Password_1@IP_Address_1/ods?charset=gbk")AdventureDw = create_engine("mysql://UserName_2:Password_2@IP_Address_2/dw?charset=gbk")# 读取ods_sales_orders表（订单明细表）SQLquery1 = """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) """OdsSaleOrder = pd.read_sql_query(SQLquery1, con=AdventureOds)# 读取ods_customer表（每日新增用户表）SQLquery2 = """select customer_key, chinese_territory, chinese_province, chinese_city from ods_customer"""OdsCustomer = pd.read_sql_query(SQLquery2, con=AdventureOds)# 读取dim_date_df表（日期维度表）SQLquery3 = """select create_date, is_current_year, is_last_year, is_yesterday, is_today, is_current_month, is_current_quarter from dim_date_df"""OdsDimDate = pd.read_sql_query(SQLquery3, con=AdventureOds)# 数据加工SaleOrderCustomer = pd.merge(OdsSaleOrder, OdsCustomer, 'left', 'customer_key')SaleOrderCustomerP = SaleOrderCustomer.pivot_table(index=['create_date', 'english_product_name', 'cpzl_zw', 'cplb_zw', 'chinese_territory', 'chinese_province', 'chinese_city'], values=['sales_order_key', 'customer_key', 'unit_price'], aggfunc={'sales_order_key': pd.Series.nunique, 'customer_key': pd.Series.nunique, 'unit_price': sum} )SaleOrderCustomerP = SaleOrderCustomerP.reset_index()SaleOrderCustomerP.rename(columns={'customer_key':'SumCustomer', 'sales_order_key':'SumOrder', 'unit_price':'SumAmount'}, inplace=True)dw_customer_order = pd.merge(SaleOrderCustomerP, OdsDimDate, 'left', 'create_date')# 加工好的dw_customer_order表(时间_地区_产品聚合表)导出至数据仓库try: pd.read_sql_query('Truncate table dw_customer_order', con=AdventureDw)except Exception as e: print('旧表删除Error: %s' %e)dw_customer_order.to_sql('dw_customer_order', con=AdventureDw, if_exists='replace', index=False)

3 ETL效率提升

ETL过程处理的数据量很大时，需要等待很长一段时间，通过一些方法可以提高ETL的效率，例如在mysql追加索引优化查询速度、在python存储至mysql时增添多进程。

3.1 追加索引

mysql索引：搜索过程不再遍历整张表，而是根据索引搜寻，索引一般是表中元素的前缀。例如，where条件筛选'2019-1'，那么索引会直接排除掉['2020','2018','2021']等这些年份结果，直接匹配前缀相符合的内容，索引的长度就是前缀的长度。下面举例说明。

给ods_sales_orders表创建日期索引前，查询耗时 3.806 ms。

# 创建日期索引前，查询日期为‘2019-03-01’订单明细select * from ods_sales_orders where create_date='2019-03-01';给ods_sales_orders表创建日期索引后，查询耗时 0.168 ms，查询速度提高 21.65 倍。可见，给表追加索引可以大大提高查询速度，提升ETL效率。

# 创建日期索引后，查询日期为‘2019-03-01’订单明细create index date_index on ods_sales_orders(create_date(7));select * from ods_sales_orders where create_date='2019-03-01';索引的工作原理：表中存在索引时，查询语句不再遍历列的所有元素，而是先遍历索引，再遍历前缀与索引相同的元素，遍历复杂度降低，从而提高查询速度。从explain结果可以看出，创建日期索引date_index后，查询首先遍历索引。

# 使用explain展示查询过程explain select * from ods_sales_orders where create_date='2019-03-01';3.2 增添多进程

进程：程序在计算机上的一次执行活动， 操作系统进行资源分配的单位。

多进程：多个程序同时在计算机上执行活动，以此实现充分调用资源目的。

除了使用索引提高mysql的查询速度，还可以利用多进程库提高python的执行效率，多进程并发可以大大减少读取数据库、写入数据库、数据加工的时间。示例：

import multiprocessing # 导入多进程库def runtask(): # dw_order_by_day.py # dw_amount_diff.py # dw_customer_order.pydef callBackTask(arg): # 回调函数必须要有一个形参，否则将报错 print("执行回调函数",arg)if __name__ == "__main__": pool = multiprocessing.Pool(5) # 设置进程池最大同时执行进程数 for index in range(20): pool.apply_async(func=runtask,callback=callBackTask) # 并行的,有回调方法 # pool.apply(func=runtask,) # 串行的，无回调函数 pool.close() # 关闭进程池 pool.join() # 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束multiprocessing库中的Pool就是进程池，进程池能够管理一定的进程，当有空闲进程时，则利用空闲进程完成任务，直到所有任务完成为止。

4 自动化部署

数据库每天都会更新，每天的数据报表也是需要更新的，那么可以借助自动化代码进行每天更新。示例：

import schedule # 定时执行模块import timeimport datetimeimport os # 命令窗口交互模块import requests​def job1(): """ dw_order_by_day 每日环比表 """ print('Job1:每天8:00执行一次') print('Job1-startTime:%s' % (datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) os.system( "/home/anaconda3/bin/python3 /home/******/adventure/dw_order_by_day.py >> /home/******/adventure/logs/dw_order_by_day_schedule.log 2>&1 &") time.sleep(20) print('Job1-endTime:%s' % (datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) print('------------------------------------------------------------------------') if __name__ == '__main__': schedule.every().day.at('08:00').do(job1) while True: schedule.run_pending() time.sleep(10) print("wait", datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))将这个文件挂在linux系统的后台，每天早上8点，自动执行，定时更新；同时将代码运行情况写入dw_order_by_day_schedule.log文件。

5 Dashboard设计

每天早上8点，数据自动更新后，Dashboard也会随之更新，方便监控日常指标。