介紹

機(jī)器學(xué)習(xí) （ML） 在數(shù)據(jù)驅(qū)動(dòng)的決策中越來越重要，因此使用現(xiàn)代工具和技術(shù)來簡(jiǎn)化機(jī)器學(xué)習(xí)工作流程非常重要。這就是 dbt 和 fal 可以發(fā)揮作用的地方 - 它們一起使以可擴(kuò)展和可重現(xiàn)的方式管理和部署機(jī)器學(xué)習(xí)模型變得容易。在這篇博文中，我們將引導(dǎo)您了解如何使用 fal 和 dbt 來訓(xùn)練和存儲(chǔ)邏輯回歸 ML 模型，對(duì)新數(shù)據(jù)進(jìn)行預(yù)測(cè)，并將這些預(yù)測(cè)存儲(chǔ)在 dbt 模型中。在這篇文章結(jié)束時(shí)，您將具備將這些工具應(yīng)用于您自己的 ML 項(xiàng)目的技能和知識(shí)。

設(shè)置

我們準(zhǔn)備了一個(gè)示例項(xiàng)目，您可以在閱讀此博客文章時(shí)使用它。它既有一個(gè)帶有一些合成數(shù)據(jù)的dbt項(xiàng)目，也有一個(gè)示例Jupyter筆記本。您可以克隆它：git clone?https://github.com/fal-ai/dbt_fal_ml_example

我們使用dbt-fal作為Python適配器。這是運(yùn)行dbt Python模型的最簡(jiǎn)單方法。該項(xiàng)目還使用 BigQuery 作為數(shù)據(jù)倉(cāng)庫(kù)。您可以編輯 require.txt 文件以適合您自己的數(shù)據(jù)倉(cāng)庫(kù)。然后，您可以通過運(yùn)行以下命令在新的 Python 環(huán)境中安裝項(xiàng)目要求：
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
讓我們?cè)陧?xiàng)目目錄中創(chuàng)建一個(gè)文件，并用必要的憑據(jù)填充它：profiles.yml
example_shop:
target: staging
outputs:
staging:
type: fal
db_profile: db
db:
type: bigquery
method: service-account-json
...
輸出應(yīng)包含數(shù)據(jù)倉(cāng)庫(kù)憑據(jù)。db
最后，讓我們啟動(dòng) Jupyter 筆記本：
jupyter notebook notebooks/Experiments.ipynb
這將在您的終端中打印出一個(gè) URL，您可以在瀏覽器中使用該 URL 并打開“Experiments.ipynb”筆記本。
我們的示例數(shù)據(jù)集模擬零售環(huán)境中的客戶訂單和訂單退貨。該數(shù)據(jù)集包含每個(gè)訂單的客戶年齡、訂單總價(jià)以及是否退回信息。

數(shù)據(jù)探索和準(zhǔn)備

在我們的筆記本中，我們首先導(dǎo)入所有必要的模塊：
import pickle
import uuid
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
from fal import FalDbt
Jupyter 筆記本允許我們運(yùn)行 shell 命令，因此我們可以運(yùn)行和：dbt seeddbt run
!dbt seed --profiles-dir ..
!dbt run --select customer_orders customer_orders_labeled --profiles-dir ..
如您所見，我們正在計(jì)算兩個(gè) dbt 模型，并且 .顧名思義，一個(gè)數(shù)據(jù)集包含標(biāo)記的數(shù)據(jù)，而另一個(gè)數(shù)據(jù)集包含未標(biāo)記的“新鮮”數(shù)據(jù)。customer_orderscustomer_orders_labeled
讓我們看一下模型。我們必須實(shí)例化：customer_orders_labeledFalDbt
faldbt = FalDbt(project_dir="..", profiles_dir="..")
現(xiàn)在我們可以將模型下載為 pandas 數(shù)據(jù)幀并打印頂部行：customer_orders_labeled
orders_df = faldbt.ref("customer_orders_labeled")
orders_df.head()
這將打印如下所示的表：
order_id customer_id total_price age return
0 210.0 488.0 187.861698 18.0 0.0
1 263.0 578.0 628.745330 18.0 0.0
2 360.0 578.0 99.154886 18.0 0.0
3 482.0 818.0 393.284591 18.0 0.0
4 594.0 656.0 339.542104 18.0 0.0
該列是數(shù)字，其中表示尚未退回訂單，表示已退回訂單。由于它是我們要預(yù)測(cè)的列的值，因此我們將此列稱為標(biāo)簽，其他列是特征。讓我們假設(shè)特征和 not 在是否退回訂單方面不起作用。這給我們留下了和.return0.01.0returnorder_idcustomer_idtotal_priceage
可視化要素和標(biāo)注之間關(guān)系的一個(gè)好方法是繪制繪圖。我們可以通過使用庫(kù)輕松做到這一點(diǎn)：

matplotlib
plot_data = orders_df.sample(frac=0.1, random_state=123)
colors = ['red' if r else 'blue' for r in plot_data['return']] # assign colors based on whether or not order was returned
plt.scatter(plot_data['age'], plot_data['total_price'], c=colors)
plt.xlabel('Age')
plt.ylabel('Total Price')
plt.show()

這是結(jié)果圖：

紅點(diǎn)對(duì)應(yīng)于已退貨的訂單。我們可以從圖中看到，左上角的訂單往往比其他訂單更頻繁地退貨。

ML 模型訓(xùn)練和評(píng)估

我們將訓(xùn)練和評(píng)估的 ML 模型類型稱為邏輯回歸。邏輯回歸適用于此問題，因?yàn)槟繕?biāo)標(biāo)簽 （） 是二進(jìn)制的（0 或 1），邏輯回歸模型可以輸出介于 0 和 1 之間的概率。在我們的例子中，邏輯回歸模型的輸出將是給定客戶的年齡和訂單總價(jià)格的訂單被退回的概率。return
我們首先將數(shù)據(jù)集拆分為訓(xùn)練集和測(cè)試集：
X_train, X_test, y_train, y_test = train_test_split(
orders_df[['age', 'total_price']],
orders_df['return'],
test_size=0.2,
random_state=42)
這會(huì)將數(shù)據(jù)集拆分為訓(xùn)練集和測(cè)試集，其中 80% 的數(shù)據(jù)用于訓(xùn)練，20% 的數(shù)據(jù)用于測(cè)試。
接下來，讓我們?cè)谟?xùn)練集上訓(xùn)練一個(gè)邏輯回歸模型。我們將使用 來自 的類對(duì)象，它是邏輯回歸的快速簡(jiǎn)單實(shí)現(xiàn)：LogisticRegressionscikit-learn
lr_model = LogisticRegression(random_state=42)
lr_model.fit(X_train, y_train)
對(duì)象的方法進(jìn)行訓(xùn)練。一旦這個(gè)單元完成計(jì)算，將使用我們的數(shù)據(jù)進(jìn)行訓(xùn)練。fitlr_modellr_model
訓(xùn)練模型后，我們可以使用以下方法評(píng)估其在測(cè)試數(shù)據(jù)上的性能：predict
Make predictions on the test data
y_pred = lr_model.predict(X_test)
Print a classification report
print(classification_report(y_test, y_pred))
這將輸出一個(gè)分類報(bào)告，其中匯總了模型的性能：
precision recall f1-score support 0.0 0.87 0.97 0.91 227 1.0 0.85 0.53 0.66 73 accuracy 0.86 300
macro avg 0.86 0.75 0.79 300
weighted avg 0.86 0.86 0.85 300
分類報(bào)告顯示，該模型在測(cè)試數(shù)據(jù)上的準(zhǔn)確率為 0.86。對(duì)于類 0（無返回），模型的精度為 87.0，對(duì)于類 0（返回），模型的精度為 85.1。0 類模型的召回率為 97.0，0 類的召回率為 53.1。該模型的 F1 分?jǐn)?shù)在 0 類為 91.0，對(duì)于 0 類為 66.1。
我們可以看到，該模型對(duì)于類 0（無返回）具有良好的精度和召回率，但對(duì)于類 1（返回）具有較低的精度和召回率。這表明該模型可能比預(yù)測(cè)將返回的訂單更能預(yù)測(cè)不會(huì)返回的訂單。盡管如此，該模型的總體準(zhǔn)確度為 0.87，表明它可以對(duì)新數(shù)據(jù)做出相當(dāng)準(zhǔn)確的預(yù)測(cè)。

使用 dbt 模型自動(dòng)執(zhí)行 ML 訓(xùn)練

將模型訓(xùn)練工作流存儲(chǔ)在 dbt 模型中，使我們能夠?qū)δＰ蛿?shù)據(jù)進(jìn)行版本控制并與其他用戶共享。我們首先在目錄中創(chuàng)建一個(gè)新的 dbt 模型：。這是一個(gè) Python 模型，我們將上面的筆記本代碼改編到模型定義中：modelsorder_return_prediction_models.py
import pickle
import uuid
import pandas as pd
import datetime
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
def model(dbt, fal):
dbt.config(materialized="table")
orders_df = dbt.ref("customer_orders_labeled")
X = orders_df[['age', 'total_price']]
y = orders_df['return']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=123)
print("Model init") lr_model = LogisticRegression(random_state=123) print("Model fitting") lr_model.fit(X_train, y_train) # Test model y_pred = lr_model.predict(X_test) print("Preparing the classification report") # Create a report and put it in a DataFrame model_name = str(uuid.uuid4()) y_test = y_test.astype(float) report = classification_report(y_test, y_pred, output_dict=True) report["model_name"] = model_name report["date"] = datetime.datetime.now() output_df = pd.DataFrame([report]) output_df = output_df.rename(columns={"0.0": "target_0", "1.0": "target_1"}) output_df.set_index("model_name") print("Saving the model") # Save model weights with open(f"ml_models/{model_name}.pkl", "wb") as f: pickle.dump(lr_model, f) return output_df
該模型獲得標(biāo)記的訂單數(shù)據(jù)并訓(xùn)練邏輯回歸模型。然后評(píng)估模型，并將報(bào)表與唯一的模型名稱一起存儲(chǔ)在數(shù)據(jù)幀中。接下來，我們將模型權(quán)重保存到本地存儲(chǔ)。您可以修改此步驟以將模型權(quán)重存儲(chǔ)在云存儲(chǔ)平臺(tái)（例如 S3）上。最后，返回輸出數(shù)據(jù)幀，因此其內(nèi)容將保留在我們的數(shù)據(jù)倉(cāng)庫(kù)中。order_return_prediction_modelslr_model
我們可以運(yùn)行這個(gè) dbt 模型：
dbt run --select order_return_prediction_models
使用存儲(chǔ)的模型進(jìn)行預(yù)測(cè)
首先，我們嘗試在Jupyter筆記本中進(jìn)行預(yù)測(cè)，然后我們創(chuàng)建一個(gè)Python模型來自動(dòng)執(zhí)行此操作。
在我們的 Jupyter 筆記本中，我們可以輕松找到最準(zhǔn)確的模型：
models_df = faldbt.ref("order_return_prediction_models")
best_model_name = models_df[
model_df.accuracy == models_df.accurary.max()
].model_name[0]
然后，我們從本地存儲(chǔ)（或云存儲(chǔ)提供商）加載此模型：
with open(f"../ml_models/{model_name}.pkl", "rb") as f:
loaded_model = pickle.load(f)
我們還加載新訂單數(shù)據(jù)并檢查其外觀：
orders_new_df = faldbt.ref("customer_orders")
orders_new_df.head()
這將打印出一個(gè)表格：
order_id customer_id total_price age
0 1037.0 981.0 193.460803 19.0
1 1027.0 940.0 680.986976 21.0
2 1039.0 123.0 952.906524 22.0
3 1043.0 860.0 545.791012 22.0
4 1046.0 316.0 887.003551 24.0
如我們所見，此數(shù)據(jù)幀的形狀與 不同 缺少列。這是我們想要預(yù)測(cè)的列。customer_orders_labeledreturn
所以，讓我們做一個(gè)預(yù)測(cè)：
predictions = loaded_model.predict(orders_new_df[["age", "total_price"]])
orders_new_df["predicted_return"] = predictions
order_new_df.head()
在上面的代碼片段中，我們首先執(zhí)行預(yù)測(cè)，然后將生成的預(yù)測(cè)附加到數(shù)據(jù)幀。這是 of 應(yīng)該的樣子：orders_new_dfheadorders_new_df
order_id customer_id total_price age predicted_return
0 1037.0 981.0 193.460803 19.0 0.0
1 1027.0 940.0 680.986976 21.0 1.0
2 1039.0 123.0 952.906524 22.0 1.0
3 1043.0 860.0 545.791012 22.0 1.0
4 1046.0 316.0 887.003551 24.0 1.0
讓我們繪制我們的預(yù)測(cè)，看看它們是否有意義：
plot_data = orders_new_df.sample(frac=0.5, random_state=123)
colors = ['red' if r else 'blue' for r in plot_data['predicted_return']]
plt.scatter(plot_data['age'], plot_data['total_price'], c=colors)
plt.xlabel('Age')
plt.ylabel('Total Price')
plt.show()
這是結(jié)果圖：

如果這些值對(duì)我們來說看起來不錯(cuò)，我們可以創(chuàng)建另一個(gè) dbt Python 模型來自動(dòng)運(yùn)行這些預(yù)測(cè)。這個(gè)新的dbt模型將首先選擇最佳的邏輯回歸模型，使用它來預(yù)測(cè)訂單是否會(huì)被退回，最后將其預(yù)測(cè)存儲(chǔ)在我們的數(shù)據(jù)倉(cāng)庫(kù)中。predicted_return
以下是我們新模型的定義：order_return_predictions.py
import pickle
def model(dbt, fal):
dbt.config(materialized="table")
models_df = dbt.ref("order_return_prediction_models")
best_model_name = models_df[
models_df.accuracy == models_df.accuracy.max()].model_name[0]
with open(f"ml_models/{best_model_name}.pkl", "rb") as f:
loaded_model = pickle.load(f)
orders_new_df = dbt.ref("customer_orders")
predictions = loaded_model.predict(orders_new_df[["age", "total_price"]])
orders_new_df["predicted_return"] = predictions
return orders_new_df
我們可以運(yùn)行這個(gè) dbt 模型：
dbt run --select order_return_predictions
結(jié)論
在這篇博文中，我們介紹了如何使用 fal 和 dbt 以可擴(kuò)展和可重現(xiàn)的方式管理和部署機(jī)器學(xué)習(xí)模型。我們使用合成購(gòu)物數(shù)據(jù)集來訓(xùn)練一個(gè)邏輯回歸模型，該模型可以預(yù)測(cè)訂單被退回的概率。我們?cè)?dbt 模型中自動(dòng)化了 ML 訓(xùn)練過程，并使用生成的 ML 模型對(duì)新數(shù)據(jù)進(jìn)行預(yù)測(cè)。最后，我們能夠?qū)⒔Y(jié)果預(yù)測(cè)存儲(chǔ)在新的dbt模型中。所有這些現(xiàn)在都可以自動(dòng)運(yùn)行。