【DSW Gallery】Gbdt-FM模型

GBDT+FM 模型是由 Gbdt+LR 延伸出来的模型。该模型利用GBDT自动进行特征筛选和组合，进而生成新的离散特征向量，再把该特征向量当做 FM 模型的输入，来产生最后的预测结果。该模型能够综合利用用户、物品和上下文等多种不同的特征，生成较为全面的推荐，在CTR点击率预估场景下使用较为广泛。
本文将介绍如何基于DSW使用 Alink 快速的构建 Gbdt+FM 模型，并且会介绍如何方便的将建立的模型部署成服务。

1. PAI-DSW 官方镜像中默认已经安装了 PyAlink，内存要求 4G 及以上。
2. 本 Notebook 的内容可以直接运行查看，不需要准备任何其他文件。

from pyalink.alink import *
useLocalEnv(2)

Use one of the following commands to start using PyAlink:
 - useLocalEnv(parallelism, flinkHome=None, config=None): run PyAlink scripts locally.
 - useRemoteEnv(host, port, parallelism, flinkHome=None, localIp="localhost", config=None): run PyAlink scripts on a Flink cluster.
 - getMLEnv(): run PyAlink scripts as PyFlink scripts, support 'flink run -py xxx.py'.
Call resetEnv() to reset environment and switch to another.
JVM listening on 127.0.0.1:58007

MLEnv(benv=<pyflink.dataset.execution_environment.ExecutionEnvironment object at 0x7fdd965a0ed0>, btenv=<pyflink.table.table_environment.BatchTableEnvironment object at 0x7fdd965a0ad0>, senv=<pyflink.datastream.stream_execution_environment.StreamExecutionEnvironment object at 0x7fdd965a0b50>, stenv=<pyflink.table.table_environment.StreamTableEnvironment object at 0x7fdd965cc050>)

PATH = "https://alink-test-data.oss-cn-hangzhou.aliyuncs.com/"
TRAIN_FILE = "adult_train.csv"
TEST_FILE = "adult_test.csv"
SCHEMA_STRING = "age bigint, workclass string, fnlwgt bigint, education string, education_num bigint,"\
    + " marital_status string, occupation string, relationship string, race string, sex string, "\
    + "capital_gain bigint, capital_loss bigint, hours_per_week bigint, native_country string, label string"
trainData = CsvSourceBatchOp() \
    .setFilePath(PATH + TRAIN_FILE) \
    .setFieldDelimiter(",") \
    .setSchemaStr(SCHEMA_STRING)
testData = CsvSourceBatchOp() \
    .setFilePath(PATH + TEST_FILE) \
    .setFieldDelimiter(",") \
    .setSchemaStr(SCHEMA_STRING)
trainData.lazyPrint(5)
BatchOperator.execute()

featureCols = ["age", "fnlwgt", "education_num", "capital_gain",
            "capital_loss", "hours_per_week", "workclass", "education", "marital_status", "occupation",
            "relationship", "race", "sex", "native_country"]
numericalCols = ["age", "fnlwgt", "education_num", "capital_gain",
            "capital_loss", "hours_per_week"]
label = "label"
vecCol = "vec"
gbdtFmPipe = Pipeline() \
    .add(
        GbdtEncoder()\
            .setLabelCol(label)\
            .setFeatureCols(featureCols)\
            .setReservedCols([label])\
            .setPredictionCol(vecCol))\
    .add(
        FmClassifier() \
            .setVectorCol(vecCol) \
            .setLabelCol(label) \
            .setReservedCols([label]) \
            .setPredictionDetailCol("detail") \
            .setPredictionCol("pred"))
model = gbdtFmPipe.fit(trainData)

result = model.transform(testData)
result.lazyPrint(5)
EvalBinaryClassBatchOp() \
    .setPredictionDetailCol("detail").setLabelCol(label).linkFrom(result) \
    .link(JsonValueBatchOp().setSelectedCol("Data") \
        .setReservedCols([]) \
        .setOutputCols(["Accuracy", "AUC", "ConfusionMatrix"]) \
        .setJsonPath(["$.Accuracy", "$.AUC", "ConfusionMatrix"])).print()

gbdtLrPipe = Pipeline() \
    .add(
        GbdtEncoder()\
            .setLabelCol(label)\
            .setFeatureCols(featureCols)\
            .setReservedCols([label])\
            .setPredictionCol(vecCol))\
    .add(
        LogisticRegression() \
            .setVectorCol(vecCol) \
            .setLabelCol(label) \
            .setReservedCols([label]) \
            .setPredictionDetailCol("detail") \
            .setPredictionCol("pred"))
lrModel = gbdtLrPipe.fit(trainData)
resultLr = lrModel.transform(testData)
resultFm = model.transform(testData)
EvalBinaryClassBatchOp() \
    .setPredictionDetailCol("detail").setLabelCol(label).linkFrom(resultLr) \
    .link(JsonValueBatchOp().setSelectedCol("Data") \
        .setReservedCols([]) \
        .setOutputCols(["Accuracy", "AUC", "ConfusionMatrix"]) \
        .setJsonPath(["$.Accuracy", "$.AUC", "ConfusionMatrix"])).print()
EvalBinaryClassBatchOp() \
    .setPredictionDetailCol("detail").setLabelCol(label).linkFrom(resultFm) \
    .link(JsonValueBatchOp().setSelectedCol("Data") \
        .setReservedCols([]) \
        .setOutputCols(["Accuracy", "AUC", "ConfusionMatrix"]) \
        .setJsonPath(["$.Accuracy", "$.AUC", "ConfusionMatrix"])).print()

  fs = OssFileSystem("3.4.1", "oss-cn-hangzhou-zmf.aliyuncs.com", "name", "************", "**********")
  filePath = FilePath("/model/gbdt_fm_model.ak", fs)

  AkSinkBatchOp().setFilePath(filePath).setOverwriteSink(True)

modelData = model.save();
filePath = "/tmp/gbdt_fm_model.ak"
# 可以将模型文件写出到OSS，这样可以直接部署到EAS，需要一个OSS的idkey。此处直接写出到/tmp 目录下
# fs = OssFileSystem("3.4.1", "oss-cn-hangzhou-zmf.aliyuncs.com", "name", "************", "**********")
# filePath = FilePath("/model/gbdt_fm_model.ak", fs)
modelData.link(AkSinkBatchOp().setFilePath(filePath).setOverwriteSink(True));
BatchOperator.execute();

model = PipelineModel.load(filePath)
result = model.transform(testData).lazyPrint(5)
BatchOperator.execute()

!./eascmd64 -i {EAS AccessKeyId} -k {EAS AccessKeySecret} -e pai-eas.cn-beijing.aliyuncs.com create config.json