使用pgml搭建使用者流失預測任務流-雲原生資料倉儲AnalyticDB-阿里雲

本文將結合電商情境下的使用者流失預測任務，介紹在AnalyticDB PostgreSQL 7.0版中如何使用pgml搭建離線和即時預測任務流。

前提條件

核心版本為v7.1.1.0及以上的AnalyticDB PostgreSQL 7.0版執行個體。
執行個體資源類型為儲存彈性模式。
已經安裝pgml外掛程式。
說明
如果您已安裝，在資料庫的Schema列表中可以看到pgml。如未安裝請提交工單，聯絡支援人員協助安裝（需要重啟執行個體）。如有卸載外掛程式需求，也請提交工單聯絡支援人員協助卸載。

背景介紹

pgml的設計理念是讓模型更靠近資料。AI/ML功能將模型載入到PostgreSQL後端進程，以UDF的形式對其進行訓練、微調、推理。訓練、微調和推理後的模型儲存在heap表中，無需單獨設計高可用或高可靠方案，營運簡單，方便使用者使用。pgml得益於存計算一體化，減少資料轉送損耗，能夠高效完成模型的訓練及服務部署。AI/ML訓練和推理流程如下。

資料介紹和分析

資料集介紹

本文所使用的資料集為Ecommerce Customer Churn Analysis and Prediction，其中包含若干客戶歷史行為資料，且已標註客戶是否在未來流失的標籤。分析和預測客戶流失可以協助企業制定更有效策略以提升客戶留存率。資料集中包含了以下欄位。

欄位名稱	描述
CustomerID	客戶唯一標識ID。
Churn	客戶流失標誌。
Tenure	客戶使用時間長度。
PreferredLoginDevice	客戶的首選登入裝置。
CityTier	客戶所在城市類別。
WarehouseToHome	從倉庫到客戶家的距離。
PreferredPaymentMode	客戶的首選付款條件。
Gender	客戶性別。
HourSpendOnApp	客戶在行動裝置 App或網站上花費的時間（單位：小時）。
NumberOfDeviceRegistered	註冊在客戶名下的裝置總數。
PreferedOrderCat	上個月客戶的首選訂單類別。
SatisfactionScore	客戶對服務的滿意度評分。
MaritalStatus	客戶婚姻狀況。
NumberOfAddress	特定客戶添加的地址總數。
Complain	客戶上月是否投訴。
OrderAmountHikeFromlastYear	與去年相比，客戶訂單金額的增長率。
CouponUsed	客戶上個月使用的優惠券總數。
OrderCount	客戶上個月的訂單總數。
DaySinceLastOrder	客戶最後一次訂單距離今天的天數。
CashbackAmount	客戶上個月的返現金額。

資料匯入

建立資料表。

CREATE TABLE raw_data_table (
    CustomerID INTEGER,
    Churn INTEGER,
    Tenure FLOAT,
    PreferredLoginDevice TEXT,
    CityTier INTEGER,
    WarehouseToHome FLOAT,
    PreferredPaymentMode TEXT,
    Gender TEXT,
    HourSpendOnApp FLOAT,
    NumberOfDeviceRegistered INTEGER,
    PreferedOrderCat TEXT,
    SatisfactionScore INTEGER,
    MaritalStatus TEXT,
    NumberOfAddress INTEGER,
    Complain INTEGER,
    OrderAmountHikeFromlastYear FLOAT,
    CouponUsed FLOAT,
    OrderCount FLOAT,
    DaySinceLastOrder FLOAT,
    CashbackAmount FLOAT
);

下載資料集，並使用COPY命令匯入CSV格式的資料來源，請根據實際情況填寫路徑。
```
COPY raw_data_table FROM '/path/to/dataset.csv' DELIMITER ',' CSV HEADER;
```
說明
此處使用psql工具進行資料匯入。若您使用其他SDK，可參照相應文檔使用COPY或INSERT方式進行匯入。

資料分析

檢查資料集空值分布。

DO $$
DECLARE
    r RECORD;
    SQL TEXT := '';
BEGIN
    FOR r IN 
        SELECT column_name 
        FROM information_schema.columns 
        WHERE table_name = 'raw_data_table'
    LOOP
        SQL := SQL || 
            'SELECT ''' || r.column_name || ''' AS column_name, COUNT(*) FILTER (WHERE ' || r.column_name || ' IS NULL) AS null_count FROM raw_data_table UNION ALL ';
    END LOOP;

    SQL := LEFT(SQL, length(SQL) - 11); 
    
    FOR r IN EXECUTE SQL LOOP
        RAISE NOTICE 'Column: %, Null Count: %', r.column_name, r.null_count;
    END LOOP;
END $$;

樣本結果如下。

NOTICE:  Column: customerid, Null Count: 0
NOTICE:  Column: churn, Null Count: 0
NOTICE:  Column: tenure, Null Count: 264
NOTICE:  Column: preferredlogindevice, Null Count: 0
NOTICE:  Column: citytier, Null Count: 0
NOTICE:  Column: warehousetohome, Null Count: 251
NOTICE:  Column: preferredpaymentmode, Null Count: 0
NOTICE:  Column: gender, Null Count: 0
NOTICE:  Column: hourspendonapp, Null Count: 255
NOTICE:  Column: numberofdeviceregistered, Null Count: 0
NOTICE:  Column: preferedordercat, Null Count: 0
NOTICE:  Column: satisfactionscore, Null Count: 0
NOTICE:  Column: maritalstatus, Null Count: 0
NOTICE:  Column: numberofaddress, Null Count: 0
NOTICE:  Column: complain, Null Count: 0
NOTICE:  Column: orderamounthikefromlastyear, Null Count: 265
NOTICE:  Column: couponused, Null Count: 256
NOTICE:  Column: ordercount, Null Count: 258
NOTICE:  Column: daysincelastorder, Null Count: 307
NOTICE:  Column: cashbackamount, Null Count: 0

對於含空值的欄位，需檢查其語義及資料分布，以便在後續訓練時確認預先處理填充策略和特徵工程。檢查詳情如下。

建立分析函數以檢查欄位是否含空值。

CREATE OR REPLACE FUNCTION print_column_statistics(table_name TEXT, column_name TEXT)
RETURNS VOID AS $$
DECLARE
    SQL TEXT;
    distinct_count INTEGER;
    min_value NUMERIC;
    max_value NUMERIC;
    avg_value NUMERIC;
    median_value NUMERIC;
    r RECORD;
BEGIN
    SQL := 'SELECT 
                COUNT(DISTINCT ' || column_name || ') AS distinct_count,
                MIN(' || column_name || ') AS min_value,
                MAX(' || column_name || ') AS max_value,
                AVG(' || column_name || ') AS avg_value,
                PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY ' || column_name || ') AS median_value
            FROM ' || table_name;

    EXECUTE SQL INTO r;

    distinct_count := r.distinct_count;
    min_value := r.min_value;
    max_value := r.max_value;
    avg_value := r.avg_value;
    median_value := r.median_value;

    RAISE NOTICE 'Distinct Count: %', distinct_count;

    IF distinct_count < 40 THEN
        SQL := 'SELECT ' || column_name || ' AS col, COUNT(*) AS count FROM ' || table_name || 
               ' GROUP BY ' || column_name || ' ORDER BY count DESC';

        FOR r IN EXECUTE SQL LOOP
            RAISE NOTICE '%: %', r.col, r.count;
        END LOOP;
    END IF;

    RAISE NOTICE 'Min Value: %, Max Value: %, Avg Value: %, Median Value: %', 
                 min_value, max_value, avg_value, median_value;
END;
$$ LANGUAGE plpgsql;

使用該函數檢查欄位中是否含空值。

SELECT print_column_statistics('raw_data_table', 'tenure');

檢查結果如下。

NOTICE:  Distinct Count: 36
NOTICE:  1: 690
NOTICE:  0: 508
NOTICE:  <NULL>: 264
NOTICE:  8: 263
NOTICE:  9: 247
NOTICE:  7: 221
NOTICE:  10: 213
NOTICE:  5: 204
...
NOTICE:  Min Value: 0, Max Value: 61, Avg Value: 10.1898993663809, Median Value: 9

模型訓練

資料預先處理

從上述分析步驟可知，本次資料集中Tenure、WareHousetohome、HourSpendOnApp、OrderAmountHikeFromLastYear、CouponUsed、OrderCount及DaySinceLastOrder欄位存在空值。為了避免影響模型的效能，需對空值進行資料處理。以下是各欄位中的資料分析及空值的處理方法。

Tenure欄位中，正常值為正偏分布，使用中位元填充空值。
WareHousetohome欄位中，距離的分布存在極端值（例如，距離倉庫比較遠的客戶），使用中位元填充使資料分布更集中。
HourSpendOnApp欄位中，分布相對對稱，可使用均值填充。
OrderAmountHikeFromLastYear欄位中，總體分布平穩，可使用均值填充。
CouponUsed欄位中，推測缺失部分為未使用過優惠券，使用零值填充。
OrderCount欄位中，推測缺失部分為未下單，使用零值填充。
DaySinceLastOrder中，空值往往表示長時間未活躍，使用該欄位中的最大值進行填充。

結合機器學習使用文檔，可以得到預先處理參數如下。

{
  "tenure": {"impute": "median"},
  "warehousetohome": {"impute": "median"},
  "hourspendonapp": {"impute": "mean"},
  "orderamounthikefromlastyear": {"impute": "mean"},
  "couponused": {"impute": "zero"},
  "ordercount": {"impute": "zero"},
  "daysincelastorder": {"impute": "max"}
}

說明

CityTier和Complain欄位類型為INTEGER，但實際上為標籤類型，數值並無大小之分。可以考慮將其轉換為TEXT類型，並使用one-hot等方式進行編碼。

建立訓練視圖

根據上述的資料處理結果，可建立對應的視圖以避免對未經處理資料表的物理修改，同時方便後續添加更多的特徵。

CREATE OR REPLACE VIEW train_data_view AS
SELECT
  Churn::TEXT,
  Tenure,
  PreferredLoginDevice,
  CityTier::TEXT,
  WarehouseToHome,
  PreferredPaymentMode,
  Gender,
  HourSpendOnApp,
  NumberOfDeviceRegistered,
  PreferedOrderCat,
  SatisfactionScore,
  MaritalStatus,
  NumberOfAddress,
  Complain::TEXT,
  OrderAmountHikeFromlastYear,
  CouponUsed,
  OrderCount,
  DaySinceLastOrder,
  CashbackAmount
FROM
  raw_data_table;

特徵工程

特徵工程是機器學習和資料採礦中一項關鍵的步驟。該步驟可對未經處理資料進行處理和轉換、提供額外的資訊輸入、實現模型的收斂、提升整體效能等。

特徵名稱	描述
AvgCashbkPerOrder	訂單平均返現金額。計算方式：CashbackAmount/OrderCount。
AvgHourSpendPerOrder	訂單平均瀏覽時間。計算方式：HourSpendOnApp/OrderCount。
CouponUsedPerOrder	訂單平均優惠券使用數。計算方式：CouponUsed/OrderCount。
LogCashbackAmount	返現金額對數化處理。計算方式：log(1+LogCashbackAmount)。

根據上述新增特徵，可重新建立視圖。

CREATE OR REPLACE VIEW train_data_view AS
SELECT
  Churn::TEXT,
  Tenure,
  PreferredLoginDevice,
  CityTier::TEXT,
  WarehouseToHome,
  PreferredPaymentMode,
  Gender,
  HourSpendOnApp,
  NumberOfDeviceRegistered,
  PreferedOrderCat,
  SatisfactionScore,
  MaritalStatus,
  NumberOfAddress,
  Complain::TEXT,
  OrderAmountHikeFromlastYear,
  CouponUsed,
  OrderCount,
  DaySinceLastOrder,
  CashbackAmount,
  CashbackAmount/OrderCount AS AvgCashbkPerOrder,
  HourSpendOnApp/OrderCount AS AvgHourSpendPerOrder,
  CouponUsed/OrderCount AS CouponUsedPerOrder,
  log(1+CashbackAmount) AS LogCashbackAmount
FROM
  raw_data_table;

模型訓練與選擇

通過pgml.train介面，可以使用不同模型快速擬合資料，並驗證其效果。以下列出XGBoost模型和bagging模型的擬合過程和輸出結果，其餘的模型由於篇幅限制在此省略。

使用XGBoost模型擬合資料。

SELECT * FROM pgml.train(
    project_name => 'Customer Churn Prediction Project', --訓練名稱
    task => 'classification', --任務類型
    relation_name => 'train_data_view', -- 資料來源
    y_column_name => 'churn', -- 預測類別欄位名
    preprocess => '{
            "tenure": {"impute": "median"},
            "warehousetohome": {"impute": "median"},
            "hourspendonapp": {"impute": "mean"},
            "orderamounthikefromlastyear": {"impute": "mean"},
            "couponused": {"impute": "zero"},
            "ordercount": {"impute": "zero"},
            "daysincelastorder": {"impute": "max"},
            "avgcashbkperorder": {"impute": "zero"},
            "avghourspendperorder": {"impute": "zero"},
            "couponusedperorder": {"impute": "zero"},
            "logcashbackamount": {"impute": "min"}
        }', -- 預先處理方法
    algorithm => 'xgboost', -- 模型類型
    runtime => 'python', -- 指定python運行環境
    test_size => 0.2 -- 測試集比例
);

輸出擬合指標如下。

-- 輸出擬合指標：
-- {f1": 0.9543147, "precision": 0.96907216, "recall": 0.94, "accuracy": 0.9840142, ...}

使用bagging模型擬合資料。

-- bagging regression
SELECT * FROM pgml.train(
    project_name => 'Customer Churn Prediction Project', --訓練名稱
    task => 'classification', --任務類型
    relation_name => 'train_data_view', -- 資料來源
    y_column_name => 'churn', -- 預測類別欄位名
    preprocess => '{
            "tenure": {"impute": "median"},
            "warehousetohome": {"impute": "median"},
            "hourspendonapp": {"impute": "mean"},
            "orderamounthikefromlastyear": {"impute": "mean"},
            "couponused": {"impute": "zero"},
            "ordercount": {"impute": "zero"},
            "daysincelastorder": {"impute": "max"},
            "avgcashbkperorder": {"impute": "zero"},
            "avghourspendperorder": {"impute": "zero"},
            "couponusedperorder": {"impute": "zero"},
            "logcashbackamount": {"impute": "min"}
        }', -- 預先處理方法
    algorithm => 'bagging', -- 模型類型
    runtime => 'python', -- 指定python運行環境
    test_size => 0.2 -- 測試集比例
);

輸出擬合指標如下。

-- 輸出擬合指標：
-- {"f1": 0.9270833, "precision": 0.96216214, "recall": 0.89447236}

通過替換algorithm參數，可驗證不同模型在該資料集上的擬合能力，支援的模型列表請參見參數列表。通過分析測試集的F1指標可知，XGBoost模型表現優於其他模型，因此本次選擇該模型作為後續操作的基準。

使用網格參數搜尋策略篩選出最合適的模型超參數，並使用5折交叉驗證增強結論可信度，搜尋策略的超參數有以下三個。

超參數名稱	描述
n_estimators	要構建樹的數量。增加樹的數量通常可以提高模型效能，但也會增加計算成本。取值範圍不定，例如設定範圍從100到500五個取值，以尋找最佳的平衡點。
eta	學習率。決定每棵樹對最終預測的貢獻程度。較小的學習率會使訓練過程更穩定，但可能需要更多的樹（`n_estimators`）來收斂。取值範圍不定，例如可設定0.05、0.1 和 0.2，以在訓練穩定性和效率之間找到合適的平衡點。
max_depth	每棵樹的最大深度。較高的深度能夠捕捉更多的特徵互動，但也容易導致過度學習。取值範圍不定，例如設定為深度16和32。

SELECT * FROM pgml.train(
    project_name => 'Customer Churn Prediction Project', --訓練名稱
    task => 'classification', --任務類型
    relation_name => 'train_data_view', -- 資料來源
    y_column_name => 'churn', -- 預測類別欄位名
    preprocess => '{
            "tenure": {"impute": "median"},
            "warehousetohome": {"impute": "median"},
            "hourspendonapp": {"impute": "mean"},
            "orderamounthikefromlastyear": {"impute": "mean"},
            "couponused": {"impute": "zero"},
            "ordercount": {"impute": "zero"},
            "daysincelastorder": {"impute": "max"},
            "avgcashbkperorder": {"impute": "zero"},
            "avghourspendperorder": {"impute": "zero"},
            "couponusedperorder": {"impute": "zero"},
            "logcashbackamount": {"impute": "min"}
        }', -- 預先處理方法
    algorithm => 'xgboost', -- 模型類型
    search_args => '{ "cv": 5 }', -- 啟用5折交叉驗證
    SEARCH => 'grid', -- 網格搜尋策略
    search_params => '{
        "max_depth": [4, 6, 8, 16], 
        "n_estimators": [100, 200, 300, 400, 500, 1000, 2000],
        "eta": [0.05, 0.1, 0.2]
    }',
    hyperparams => '{
        "nthread": 16,
        "alpha": 0,
        "lambda": 1
    }',
    runtime => 'python', -- 指定python運行環境
    test_size => 0.2 -- 測試集比例
);

樣本結果如下。

-- 搜尋結果
-- ... （省略詳細輸出）
INFO:  Best Hyperparams: {
  "alpha": 0,
  "lambda": 1,
  "nthread": 16,
  "eta": 0.1,
  "max_depth": 6,
  "n_estimators": 1000
}
INFO:  Best f1 Metrics: Number(0.9874088168144226)

從結果可得出，當超參數取值為{"eta": 0.2, "max_depth": 16, "n_estimators": 400}時，模型擬合能力最強。由於驗證時使用了K-fold交叉驗證（即search_args => '{ "cv": 5 }），因此模型所使用的訓練資料為資料集中的80%。

使用最佳超參數在全量資料上訓練模型並驗證結果。

SELECT * FROM pgml.train(
    project_name => 'Customer Churn Prediction Project', --訓練名稱
    task => 'classification', --任務類型
    relation_name => 'train_data_view', -- 資料來源
    y_column_name => 'churn', -- 預測類別欄位名
    preprocess => '{
            "tenure": {"impute": "median"},
            "warehousetohome": {"impute": "median"},
            "hourspendonapp": {"impute": "mean"},
            "orderamounthikefromlastyear": {"impute": "mean"},
            "couponused": {"impute": "zero"},
            "ordercount": {"impute": "zero"},
            "daysincelastorder": {"impute": "max"},
            "avgcashbkperorder": {"impute": "zero"},
            "avghourspendperorder": {"impute": "zero"},
            "couponusedperorder": {"impute": "zero"},
            "logcashbackamount": {"impute": "min"}
        }', -- 預先處理方法
    algorithm => 'xgboost', -- 模型類型
    hyperparams => '{
        "max_depth": 6,
        "n_estimators": 1000,
        "eta": 0.1,
        "nthread": 16,
        "alpha": 0,
        "lambda": 1
    }',
    runtime => 'python', -- 指定python運行環境
    test_size => 0.2 -- 測試集比例
);

樣本結果如下。

-- 輸出結果
INFO:  Training Model { id: 170, task: classification, algorithm: xgboost, runtime: python }
INFO:  Hyperparameter searches: 1, cross validation folds: 1
INFO:  Hyperparams: {
  "eta": 0.1,
  "alpha": 0,
  "lambda": 1,
  "nthread": 16,
  "max_depth": 6,
  "n_estimators": 1000
}
INFO:  Metrics: {"roc_auc": 0.9751001, "log_loss": 0.19821791, "f1": 0.99258476, "precision": 0.9936373, "recall": 0.9915344, "accuracy": 0.9875666, "mcc": 0.95414394, "fit_time": 0.9980099, "score_time": 0.0085158}
INFO:  Comparing to deployed model f1: Some(0.9874088168144226)
INFO:  Deploying model id: 170
              project              |      task      | algorithm | deployed
-----------------------------------+----------------+-----------+----------
 Customer Churn Prediction Project | classification | xgboost   | t

在測試集上模型效能可達到F1=0.99258476。

模型部署

選取部署模型

預設情況下，pgml會自動部署project內訓練過程中F1值最高的模型（針對分類任務），可以使用pgml.deployments表檢查當前的部署情況。

SELECT d.id, d.project_id, d.model_id, p.name, p.task FROM pgml.deployments d 
JOIN pgml.projects p on d.project_id = p.id;

樣本結果如下。

 id | project_id | model_id |               name                |      task
----+------------+----------+-----------------------------------+----------------
 61 |          2 |      170 | Customer Churn Prediction Project | classification

如果您需要指定部署其他模型，請參見部署。

模型使用

線上推理

線上推理適用於需要即時互動式響應的情境。例如資料分析師進行個案分析時，需要根據使用者的歷史行為立即返回預測結果。

-- 單條線上推理
SELECT pgml.predict('Customer Churn Prediction Project', 
( 4, 'Mobile Phone'::TEXT, 3, 6, 
'Debit Card'::TEXT, 'Female'::TEXT, 3, 3, 
'Laptop & Accessory'::TEXT, 2, 
'Single'::TEXT, 9 ,
'1'::TEXT, 11, 1, 1, 5, 159.93, 
159.93, 3, 1, 2.206637011283536
));

樣本結果如下。

-- 預測輸出
 predict
---------
       0
(1 row)

離線推理

離線推理適合需要處理資料量較大，吞吐率優先於RT的情境，能夠提高計算資源的使用率。

-- 建立視圖
CREATE OR REPLACE VIEW predict_data_view AS
SELECT
  CustomerID,
  Churn::TEXT,
  Tenure,
  PreferredLoginDevice,
  CityTier::TEXT,
  WarehouseToHome,
  PreferredPaymentMode,
  Gender,
  HourSpendOnApp,
  NumberOfDeviceRegistered,
  PreferedOrderCat,
  SatisfactionScore,
  MaritalStatus,
  NumberOfAddress,
  Complain::TEXT,
  OrderAmountHikeFromlastYear,
  CouponUsed,
  OrderCount,
  DaySinceLastOrder,
  CashbackAmount,
  CashbackAmount/OrderCount AS AvgCashbkPerOrder,
  HourSpendOnApp/OrderCount AS AvgHourSpendPerOrder,
  CouponUsed/OrderCount AS CouponUsedPerOrder,
  log(1+CashbackAmount) AS LogCashbackAmount
FROM
  raw_data_table;

-- 批量離線推理
SELECT CustomerID, pgml.predict('Customer Churn Prediction Project', (
  "tenure",
  "preferredlogindevice",
  "citytier",
  "warehousetohome",
  "preferredpaymentmode",
  "gender",
  "hourspendonapp",
  "numberofdeviceregistered",
  "preferedordercat",
  "satisfactionscore",
  "maritalstatus",
  "numberofaddress",
  "complain",
  "orderamounthikefromlastyear",
  "couponused",
  "ordercount",
  "daysincelastorder",
  "cashbackamount",
  "avgcashbkperorder",
  "avghourspendperorder",
  "couponusedperorder",
  "logcashbackamount"
)) FROM predict_data_view limit 20;

樣本結果如下。

-- 預測結果
 customerid | predict
------------+---------
      50005 |       0
      50009 |       0
      50012 |       0
      50013 |       0
      50019 |       0
      50020 |       0
      50022 |       0
      50023 |       0
      50026 |       0
      50031 |       1
      50039 |       1
      50040 |       0
      50043 |       1
      50045 |       1
      50047 |       0
      50048 |       1
      50050 |       1
      50051 |       1
      50052 |       1
      50053 |       0
(20 rows)