【DSW Gallery】Gbdt-LR模型

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

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:57792

MLEnv(benv=<pyflink.dataset.execution_environment.ExecutionEnvironment object at 0x7fb455ca3950>, btenv=<pyflink.table.table_environment.BatchTableEnvironment object at 0x7fb455ca3d10>, senv=<pyflink.datastream.stream_execution_environment.StreamExecutionEnvironment object at 0x7fb455ca3e90>, stenv=<pyflink.table.table_environment.StreamTableEnvironment object at 0x7fb455cc7c10>)

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"
gbdtLrPipe = Pipeline() \
    .add(
        GbdtEncoder()\
            .setLabelCol(label)\
            .setFeatureCols(featureCols)\
            .setReservedCols([label])\
            .setPredictionCol(vecCol))\
    .add(
        LogisticRegression() \
            .setVectorCol(vecCol) \
            .setLabelCol(label) \
            .setReservedCols([label]) \
            .setPredictionDetailCol("detail") \
            .setPredictionCol("pred"))
model = gbdtLrPipe.fit(trainData)

result = model.transform(testData)
EvalBinaryClassBatchOp() \
    .setPredictionDetailCol("detail").setLabelCol(label).linkFrom(result) \
    .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_lr_model.ak", fs)

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

modelData = model.save();
filePath = "/tmp/gbdt_lr_model.ak"
# 可以将模型文件写出到OSS，这样可以直接部署到EAS，需要一个OSS的idkey。此处直接写出到/tmp 目录下
# fs = OssFileSystem("3.4.1", "oss-cn-hangzhou-zmf.aliyuncs.com", "name", "************", "**********")
# filePath = FilePath("/model/gbdt_lr_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