CDA數據分析師 出品

作者：CDA教研組

編輯：Mika

案例介紹

背景：以某大型電商平臺的用戶行為數據為數據集，使用大數據處理技術分析海量數據下的用戶行為特征，并通過建立邏輯回歸模型、隨機森林對用戶行為做出預測;

案例思路:

• 使用大數據處理技術讀取海量數據
• 海量數據預處理
• 抽取部分數據調試模型
• 使用海量數據搭建模型

#全部行輸出
from
IPython.core.interactiveshell import InteractiveShell

InteractiveShell.ast_node_interactivity = "all"

數據字典:

U_Id:the serialized ID that represents a user

T_Id:the serialized ID that represents an item

C_Id:the serialized ID that represents the category which the corresponding item belongs to Ts:the timestamp of the behavior

Be_type:enum-type from (‘pv’, ‘buy’, ‘cart’, ‘fav’)

pv: Page view of an item's detail page, equivalent to an item click

buy: Purchase an item

cart: Add an item to shopping cart
fav: Favor an item

讀取數據

這里關鍵是使用dask庫來處理海量數據，它的大多數操作的運行速度比常規pandas等庫快十倍左右。

pandas在分析結構化數據方面非常的流行和強大，但是它最大的限制就在于設計時沒有考慮到可伸縮性。pandas特別適合處理小型結構化數據，并且經過高度優化，可以對存儲在內存中的數據執行快速高 效的操作。然而隨著數據量的大幅度增加，單機肯定會讀取不下的，通過集群的方式來處理是最好的選 擇。這就是Dask DataFrame API發揮作用的地方:通過為pandas提供一個包裝器，可以智能的將巨大的DataFrame分隔成更小的片段，并將它們分散到多個worker(幀)中，并存儲在磁盤中而不是RAM中。

Dask DataFrame會被分割成多個部門，每個部分稱之為一個分區，每個分區都是一個相對較小的 DataFrame，可以分配給任意的worker，并在需要復制時維護其完整數據。具體操作就是對每個分區并 行或單獨操作(多個機器的話也可以并行)，然后再將結果合并，其實從直觀上也能推出Dask肯定是這么做的。

# 安裝庫(清華鏡像)
# pip install dask -i
https://pypi.tuna.tsinghua.edu.cn/simple
import os
import gc # 垃圾回收接口
from tqdm import tqdm # 進度條庫
import dask # 并行計算接口
from dask.diagnostics import ProgressBar
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import time
import dask.dataframe as dd # dask中的數表處理庫 import sys # 外部參數獲取接口

面對海量數據，跑完一個模塊的代碼就可以加一行gc.collect()來做內存碎片回收，Dask Dataframes與Pandas Dataframes具有相同的API

gc.collect()

42

# 加載數據
data = dd.read_csv('UserBehavior_all.csv')# 需要時可以設置blocksize=參數來手工指定劃分方法，默認是64MB(需要設置為總線的倍數，否則會放慢速度)
data.head()

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

data
Dask DataFrame Structure :

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

Dask Name: read-csv, 58 tasks

與pandas不同，這里我們僅獲取數據框的結構，而不是實際數據框。Dask已將數據幀分為幾塊加載，這些塊存在 于磁盤上，而不存在于RAM中。如果必須輸出數據幀，則首先需要將所有數據幀都放入RAM，將它們縫合在一 起，然后展示最終的數據幀。使用.compute()強迫它這樣做，否則它不.compute() 。其實dask使用了一種延遲數 據加載機制，這種延遲機制類似于python的迭代器組件，只有當需要使用數據的時候才會去真正加載數據。

# 真正加載數據 data.compute()

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

# 可視化工作進程，58個分區任務 data.visualize()

數據預處理

數據壓縮

# 查看現在的數據類型 data.dtypes

U_Id int64
T_Id int64
C_Id int64
Be_type object
Ts int64
dtype: object

# 壓縮成32位uint，無符號整型，因為交易數據沒有負數 dtypes = {
'U_Id': 'uint32',
'T_Id': 'uint32',
'C_Id': 'uint32',
'Be_type': 'object',
'Ts': 'int64'
}
data = data.astype(dtypes)

data.dtypes

U_Id uint32
T_Id uint32
C_Id uint32
Be_type object
Ts int64
dtype: object

缺失值

# 以dask接口讀取的數據，無法直接用.isnull()等pandas常用函數篩查缺失值
data.isnull()

Dask DataFrame Structure :

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

columns1 = [ 'U_Id', 'T_Id', 'C_Id', 'Be_type', 'Ts']
tmpDf1 = pd.DataFrame(columns=columns1)
tmpDf1

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

s = data["U_Id"].isna()
s.loc[s == True]

Dask Series Structure:
npartitions=58
bool ...
... ...
...
Name: U_Id, dtype: bool
Dask Name: loc-series, 348 tasks

U_Id列缺失值數目為0
T_Id列缺失值數目為0
C_Id列缺失值數目為0
Be_type列缺失值數目為0
Ts列缺失值數目為0

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

無缺失值

數據探索與可視化

這里我們使用pyecharts庫。pyecharts是一款將python與百度開源的echarts結合的數據可視化工具。新版的1.X和舊版的0.5.X版本代碼規則大 不相同，新版詳見官方文檔
https://gallery.pyecharts.org/#/README

# pip install pyecharts -i https://pypi.tuna.tsinghua.edu.cn/simple

Looking in indexes: https:
//pypi.tuna.tsinghua.edu.cn/simple
Requirement already satisfied: pyecharts in d:anacondalibsite-packages (0.1.9.4)
Requirement already satisfied: jinja2 in d:anacondalibsite-packages (from pyecharts)
(3.0.2)
Requirement already satisfied: future in d:anacondalibsite-packages (from pyecharts)
(0.18.2)
Requirement already satisfied: pillow in d:anacondalibsite-packages (from pyecharts)
(8.3.2)
Requirement already satisfied: MarkupSafe>=2.0 in d:anacondalibsite-packages (from
jinja2->pyecharts) (2.0.1)
Note: you may need to restart the kernel to use updated packages.
U_Id列缺失值數目為0 T_Id列缺失值數目為0 C_Id列缺失值數目為0 Be_type列缺失值數目為0 Ts列缺失值數目為0

WARNING: Ignoring invalid distribution -umpy (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -ip (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -umpy (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -ip (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -umpy (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -ip (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -umpy (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -ip (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -umpy (d:anacondalibsite-packages)
WARNING: Ignoring invalid distribution -ip (d:anacondalibsite-packages)

餅圖

# 例如，我們想畫一張漂亮的餅圖來看各種用戶行為的占比 data["Be_type"]

# 使用dask的時候，所有支持的原pandas的函數后面需加.compute()才能最終執行
Be_counts = data["Be_type"].value_counts().compute()
Be_counts

pv 89716264
cart 5530446
fav 2888258
buy 2015839
Name: Be_type, dtype: int64

Be_index = Be_counts.index.tolist() # 提取標簽
Be_index

['pv', 'cart', 'fav', 'buy']

Be_values = Be_counts.values.tolist() # 提取數值
Be_values

[89716264, 5530446, 2888258, 2015839]

from pyecharts import options as opts
from pyecharts.charts import Pie

#pie這個包里的數據必須傳入由元組組成的列表
c = Pie()
c.add("", [list(z) for z in zip(Be_index, Be_values)]) # zip函數的作用是將可迭代對象打包成一 個個元組，然后返回這些元組組成的列表 c.set_global_opts(title_opts=opts.TitleOpts(title="用戶行為")) # 全局參數(圖命名) c.set_series_opts(label_opts=opts.LabelOpts(formatter=": {c}"))
c.render_notebook() # 輸出到當前notebook環境
# c.render("pie_base.html") # 若需要可以將圖輸出到本機

<pyecharts.charts.basic_charts.pie.Pie at 0x1b2da75ae48>

<div id="490361952ca944fcab93351482e4b254" style="width:900px; height:500px;"></div>

漏斗圖

from pyecharts.charts import Funnel # 舊版的pyecharts不需要.charts即可import import pyecharts.options as opts
from IPython.display import Image as IMG
from pyecharts import options as opts
from pyecharts.charts import Pie

<pyecharts.charts.basic_charts.funnel.Funnel at 0x1b2939d50c8>

<div id="071b3b906c27405aaf6bc7a686e36aaa" style="width:800px; height:400px;"></div>

數據分析

時間戳轉換

dask對于時間戳的支持非常不友好

type(data)

dask.dataframe.core.DataFrame

data['Ts1']=data['Ts'].apply(lambda x: time.strftime("%Y-%m-%d %H:%M:%S",
time.localtime(x)))
data['Ts2']=data['Ts'].apply(lambda x: time.strftime("%Y-%m-%d", time.localtime(x)))
data['Ts3']=data['Ts'].apply(lambda x: time.strftime("%H:%M:%S", time.localtime(x)))

D:anacondalibsite-packagesdaskdataframecore.py:3701: UserWarning:
You did not provide metadata, so Dask is running your function on a small dataset to
guess output types. It is possible that Dask will guess incorrectly.
To provide an explicit output types or to silence this message, please provide the
`meta=` keyword, as described in the map or apply function that you are using.
Before: .apply(func)
After: .apply(func, meta=('Ts', 'object'))
warnings.warn(meta_warning(meta))

data.head(1)

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

data.dtypes

U_Id uint32
T_Id uint32
C_Id uint32
Be_type object
Ts int64
Ts1 object
Ts2 object
Ts3 object
dtype: object

抽取一部分數據來調試代碼

df = data.head(1000000)
df.head(1)

.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}

用戶流量和購買時間情況分析

用戶行為統計表

describe = df.loc[:,["U_Id","Be_type"]]
ids = pd.DataFrame(np.zeros(len(set(list(df["U_Id"])))),index=set(list(df["U_Id"])))
pv_class=describe[describe["Be_type"]=="pv"].groupby("U_Id").count()
pv_class.columns = ["pv"]
buy_class=describe[describe["Be_type"]=="buy"