淘宝用户购物行为分析

1.从https://tianchi.aliyun.com/dataset/649下载淘宝用户行为数据集

数据说明如下

1.概述
UserBehavior是阿里巴巴提供的一个淘宝用户行为数据集，用于隐式反馈推荐问题的研究。

2.介绍
文件名称 说明 包含特征
UserBehavior.csv 包含所有的用户行为数据 用户ID，商品ID，商品类目ID，行为类型，时间戳
UserBehavior.csv
本数据集包含了2017年11月25日至2017年12月3日之间，有行为的约一百万随机用户的所有行为（行为包括点击、购买、加购、喜欢）。数据集的组织形式和MovieLens-20M类似，即数据集的每一行表示一条用户行为，由用户ID、商品ID、商品类目ID、行为类型和时间戳组成，并以逗号分隔。关于数据集中每一列的详细描述如下：

 

2.使用python读取其中得1000W导入到postgres

2.1根据淘宝数据集说明在postgres创建表

CREATE TABLE IF NOT EXISTS public.userbehavior
(
userid integer,
itemid integer,
categoryid integer,
behavior character varying(50) COLLATE pg_catalog."default",
dtime bigint
)

2.2 使用python将数据样本导入数据库

import  pandas as pd

import psycopg2

from io import StringIO

from psycopg2.extras import execute_values

from sqlalchemy import create_engine

# 方法1: 使用pandas.to_sql (便捷但性能一般)

def method1_pandas_to_sql():

    dataparth=r"F:\download\UserBehavior\UserBehavior.csv"

    df=pd.read_csv(dataparth,nrows=10000000, header=None)

    engine = create_engine('postgresql://user:password@localhost:5432/UserBehaviorDB')

    df.to_sql('userbehavior', engine, if_exists='append', index=False)

# 方法2: 使用psycopg2 execute_values (性能较好)

def method2_execute_values():

    dataparth=r"F:\download\UserBehavior\UserBehavior.csv"

    df=pd.read_csv(dataparth,nrows=10000000, header=None)

    conn = psycopg2.connect(

        host="localhost",

        database="UserBehaviorDB",

        user="postgres",

        password="password"

    )

    cursor = conn.cursor()

    sql = "INSERT INTO userbehavior VALUES %s"

    execute_values(cursor, sql, df.values.tolist(), template=None, page_size=1000)

    conn.commit()

    cursor.close()

    conn.close()

 # 方法3: 使用psycopg2 COPY (高性能)

def method_copy_from():

    dataparth=r"F:\download\UserBehavior\UserBehavior.csv"

    df=pd.read_csv(dataparth,nrows=10000000, header=None)

 

    conn = psycopg2.connect(

        host="localhost",

        database="UserBehaviorDB",

        user="postgres",

        password="password"

    )

    cursor = conn.cursor()

    output = StringIO()

    df.to_csv(output, sep='\t', header=False, index=False)

    output.seek(0)

    cursor.copy_from(output, table='userbehavior')

    conn.commit()

    cursor.close()

    conn.close()

method_copy_from()

3.查看导入postgres的数据

 

 3. 数据清洗

首次清理时间异常数据

delete from userbehavior
where dtime<0

根据时间，用户分组统计数据如下：

select dtime,userid,count(1),STRING_AGG(behavior,',') AS behaviorlist,STRING_AGG(cast(itemid as varchar(10)),',') AS itemlist
from userbehavior
group by dtime,userid
having count(1)>1

 

基于同一个用户同一时间点基本不可能对大于1个商品进行两种不同的操作，因此将此类数据作为脏数据处理

select dtime,userid,count(1),STRING_AGG(behavior,',') AS behaviorlist,STRING_AGG(cast(itemid as varchar(10)),',') AS itemlist
from userbehavior
group by dtime,userid
having count(1)>1 and count(distinct behavior)>1 and count(distinct itemid)>1

 对以上脏数据进行删除

CREATE TEMPORARY TABLE temp_filtered_data AS
SELECT dtime, userid, count(1) as cnt,
STRING_AGG(behavior,',') AS behaviorlist,
STRING_AGG(cast(itemid as varchar(10)),',') AS itemlist
FROM userbehavior
GROUP BY dtime, userid
HAVING count(1)>1 AND count(distinct behavior)>1 AND count(distinct itemid)>1;

select sum(cnt) from temp_filtered_data
union all
select count(1) from temp_filtered_data a inner join userbehavior b
on a.dtime=b.dtime and a.userid=b.userid

 数据一致，对异常数据进行删除处理

delete from userbehavior
where (dtime,userid) in (select dtime,userid from temp_filtered_data)

 行为未有异常

 

-- 确认行为类型与文档相符
select distinct behavior from userbehavior
--判断关键数据是否有空值
select * from userbehavior where itemid is null or categoryid is null or dtime is null or userid is null

 

时间格式转换

select *, to_timestamp(dtime),to_timestamp(dtime) AT TIME ZONE 'UTC' AS utc_time
,to_char(to_timestamp(dtime), 'YYYY-MM-DD HH24:MI:SS')
from userbehavior limit 10

alter table public.userbehavior
add formatted_dtime timestamp

update userbehavior set formatted_dtime=to_timestamp(dtime) AT TIME ZONE 'UTC'

update耗时1分钟

PostgreSQL 的 UPDATE 操作本质上是“先删除旧行，再插入新行”，这会占用大量资源并可能导致表膨胀（即产生大量“死元组”）。因此，在处理大量数据时，应尽量避免全表更新或长时间持有锁。

如果需要update的，可根据实际情况新建一张表， 如果对业务不产生影响的可以采用update

 样本为2017年11月25日至2017年12月3日之间，对区间外数据进行删除：

 delete  from userbehavior where formatted_dtime<'2017-11-25' or  formatted_dtime>'2017-12-04'

 目前的tableau无法兼容最新版本postges,将postgres数据导入MySQL

mysql建库，建表

获取字段

behavior,categoryid,dtime,formatted_dtime,itemid,userid

postgres导入mysql  python代码如下：

import psycopg2

import pymysql

import time

def get_pg_connection():

    """建立PostgreSQL连接"""

    return psycopg2.connect(

        host="localhost",

        database="UserBehaviorDB",

        user="postgres",

        password="password",

        port="5432"

    )

def get_mysql_connection():

    """建立MySQL连接"""

    return pymysql.connect(

        host="localhost",

        database="UserBehaviorDB",

        user="root",

        password="password",

        port=3306,

        charset='utf8mb4'

    )

BATCH_SIZE = 100000  # 每批处理记录数

def migrate_data():

    # 连接 PostgreSQL

    pg_conn = get_pg_connection()

    pg_cursor = pg_conn.cursor(name='data_fetcher')  # 使用服务器端游标避免内存溢出

    pg_cursor.itersize = BATCH_SIZE

    # 连接 MySQL

    mysql_conn = get_mysql_connection()

    mysql_cursor = mysql_conn.cursor()

   

    # 禁用自动提交以提高写入效率

    mysql_conn.autocommit(False)

    # 查询语句 (根据实际表结构修改)

    select_sql = "SELECT behavior,categoryid,dtime,formatted_dtime,itemid,userid FROM userbehavior"

    insert_sql = "INSERT INTO userbehavior (behavior,categoryid,dtime,formatted_dtime,itemid,userid) VALUES (%s, %s, %s, %s, %s, %s)"

    try:

        pg_cursor.execute(select_sql)

        batch_data = []

        count = 0

        start_time = time.time()

        for row in pg_cursor:

            batch_data.append(row)

            count += 1

           

            if len(batch_data) >= BATCH_SIZE:

                mysql_cursor.executemany(insert_sql, batch_data)

                mysql_conn.commit()

                batch_data.clear()

                print(f"已迁移 {count} 条记录...")

        # 处理剩余数据

        if batch_data:

            mysql_cursor.executemany(insert_sql, batch_data)

            mysql_conn.commit()

        end_time = time.time()

        print(f"迁移完成！总共 {count} 条记录，耗时 {end_time - start_time:.2f} 秒")

    except Exception as e:

        mysql_conn.rollback()

        print(f"发生错误: {e}")

    finally:

        pg_cursor.close()

        pg_conn.close()

        mysql_cursor.close()

        mysql_conn.close()

migrate_data();

 

根据现有数据，对DW进行数据库设计，确定粒度（明确“一行数据代表什么”，数据统计的精细程度），维度（用户，商品，商品类目，行为，日期等），事实表（用户行为日志），在此不做具体描述

 4.根据已知数据使用PowerDesigner进行设计DW

在 PowerDesigner 中进行数据库设计时，通常遵循 ‌概念模型 → 逻辑模型 → 物理模型‌ 的三阶段建模流程。这种分层方式有助于从抽象业务需求逐步细化到具体数据库实现。

概念模型

‌目的‌：面向业务人员与技术人员沟通，抽象出实体及其关系，‌不依赖任何 DBMS‌。
‌核心内容‌：
实体（Entity）：如“员工”“部门”。
实体间关系：1:1、1:n、n:m。
属性（Attribute）：实体的特征，如员工姓名、出生日期。
‌特点‌：
高度抽象，无主键/外键等技术细节。
使用 E-R 图表示。

新建概念模型

 用户实体如下：

 

 

 

用户与用户行为关联如下所示：

 

根据实际情况设置对应关系

 整体概念图如下所示

右键->check model或者F4  检查模型，保证模型没有错误（不同实体属性存在相同的code或者name会报错）

 

 2.概念模型到逻辑模型

 逻辑模型

‌目的‌：在 CDM 基础上细化结构，‌独立于具体 DBMS‌，但引入技术元素如主键、外键。
‌核心内容‌：
实体转换为表（Table）。
明确主键（Identifier）、外键（Foreign Key）。
定义字段类型（如整型、字符串），但未绑定具体 DBMS 类型。
‌特点‌：
比 CDM 更结构化，仍不涉及索引、存储等物理细节。
PowerDesigner 15+ 版本引入 LDM，作为 CDM 与 PDM 的中间层 ‌

基于现有概念模型生成逻辑模型

在菜单栏选择 Tools → Generate Logical Data Model

 

 删除用户行为表多余字段，并调整关联关系

 

右键->check model或者F4  检查模型

 

逻辑模型转为物理模型

 ‌PowerDesigner中没有主键的实体确实可能导致无法正常转换为物理模型‌

 数据库选择MySQL

 

 

根据需要进行调整，在此不做赘述

 物理模型转数据库

 

 创建建表脚本

 根据生成的脚本建表

生成日期维表数据

CREATE DEFINER=`root`@`localhost` PROCEDURE `userbehaviordb`.`pro_add_date`(

IN sdate DATETIME,

IN edate DATETIME

)

BEGIN

WHILE sdate <= edate DO

insert into userbehaviordb.dim_date

select DATE_FORMAT(sdate,'%Y%m%d'),DATE_FORMAT(sdate,'%Y'),DATE_FORMAT(sdate,'%Y%m'),concat(cast(year(sdate) as char(4)),right(concat('00',cast(QUARTER(sdate) as char(2))),2));

SET sdate = DATE_ADD(sdate, INTERVAL 1 DAY);

 

END WHILE;

 

END

 填充其他维度数据

 在处理过程中发现同一个物品ID属于两个类目的情况（大概率属于正常的情况），因此需要调整物品维度表的主键，正式业务处理过程中，基于这种情况，根据需要，会在数据转换过程中重新生成一个新的itemid，在此不做赘述

insert into userbehaviordb.fact_userbehavior(userid,behavior,Itemid,item_category_id,dateid,ttime)

select userid,behavior,itemid,categoryid,date_format(formatted_dtime,'%Y%m%d'),formatted_dtime from userbehaviordb.userbehavior

 数据分析

1.总概况

统计总用户数，各行为访问量，总访问量，pv访客数，buy访客数，以及点击购买转化率的统计

select count(distinct userid) usernum

,sum(case when behavior='pv' then 1 else 0 end) pvnum

,sum(case when behavior='cart' then 1 else 0 end) cartnum

,sum(case when behavior='fav' then 1 else 0 end) favnum

,sum(case when behavior='buy' then 1 else 0 end) buynum

,count(1) lognum

,count(distinct case when behavior='pv' then userid end) pvusernum

,count(distinct case when behavior='buy' then userid end) buyusernum

,count(distinct case when behavior='buy' then userid end)/count(distinct case when behavior='pv' then userid end) '转化率'

from fact_userbehavior fu

 

 月度数据统计

select b.monthid,count(1) lognum,count(distinct case when behavior='pv' then userid end) pvusernum

,count(distinct case when behavior='buy' then userid end) buyusernum

,count(distinct case when behavior='buy' then userid end)/count(distinct case when behavior='pv' then userid end) '转化率'

from fact_userbehavior a inner join dim_date b

on a.dateid =b.dateid

group by b.monthid