paddlepaddle--paddle/test/legacy_test/dist_fleet_sparse_embedding_ctr.py

#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Distribute CTR model for test fleet api
"""

import os

import numpy as np
from test_dist_fleet_base import FleetDistRunnerBase, runtime_main

import paddle
from paddle import base


def fake_ctr_reader():
    def reader():
        for _ in range(1000):
            deep = np.random.random_integers(0, 1e10, size=16).tolist()
            wide = np.random.random_integers(0, 1e10, size=8).tolist()
            label = np.random.random_integers(0, 1, size=1).tolist()
            yield [deep, wide, label]

    return reader


class TestDistCTR2x2(FleetDistRunnerBase):
    """
    For test CTR model, using Fleet api
    """

    def net(self, args, batch_size=4, lr=0.01):
        """
        network definition

        Args:
            batch_size(int): the size of mini-batch for training
            lr(float): learning rate of training
        Returns:
            avg_cost: DenseTensor of cost.
        """
        dnn_input_dim, lr_input_dim = 10, 10

        dnn_data = paddle.static.data(
            name="dnn_data",
            shape=[-1, 1],
            dtype="int64",
        )
        lr_data = paddle.static.data(
            name="lr_data",
            shape=[-1, 1],
            dtype="int64",
        )
        label = paddle.static.data(
            name="click",
            shape=[-1, 1],
            dtype="int64",
        )

        data = [dnn_data, lr_data, label]

        if args.reader == "pyreader":
            self.reader = base.io.PyReader(
                feed_list=data,
                capacity=64,
                iterable=False,
                use_double_buffer=False,
            )

        # build dnn model
        initializer = int(os.getenv("INITIALIZER", "0"))
        inference = bool(int(os.getenv("INFERENCE", "0")))

        if initializer == 0:
            init = paddle.nn.initializer.Constant(value=0.01)
        elif initializer == 1:
            init = paddle.nn.initializer.Uniform()
        elif initializer == 2:
            init = paddle.nn.initializer.Normal()
        else:
            raise ValueError(f"error initializer code: {initializer}")

        entry = paddle.distributed.ShowClickEntry("show", "click")
        dnn_layer_dims = [128, 64, 32]
        dnn_embedding = paddle.static.nn.sparse_embedding(
            input=dnn_data,
            size=[dnn_input_dim, dnn_layer_dims[0]],
            is_test=inference,
            entry=entry,
            param_attr=base.ParamAttr(name="deep_embedding", initializer=init),
        )
        dnn_pool = paddle.static.nn.sequence_lod.sequence_pool(
            input=dnn_embedding, pool_type="sum"
        )
        dnn_out = dnn_pool
        for i, dim in enumerate(dnn_layer_dims[1:]):
            fc = paddle.static.nn.fc(
                x=dnn_out,
                size=dim,
                activation="relu",
                weight_attr=base.ParamAttr(
                    initializer=paddle.nn.initializer.Constant(value=0.01)
                ),
                name=f'dnn-fc-{i}',
            )
            dnn_out = fc

        # build lr model
        lr_embedding = paddle.static.nn.sparse_embedding(
            input=lr_data,
            size=[lr_input_dim, 1],
            is_test=inference,
            entry=entry,
            param_attr=base.ParamAttr(
                name="wide_embedding",
                initializer=paddle.nn.initializer.Constant(value=0.01),
            ),
        )

        lr_pool = paddle.static.nn.sequence_lod.sequence_pool(
            input=lr_embedding, pool_type="sum"
        )
        merge_layer = paddle.concat([dnn_out, lr_pool], axis=1)

        predict = paddle.static.nn.fc(
            x=merge_layer, size=2, activation='softmax'
        )

        acc = paddle.static.accuracy(input=predict, label=label)
        auc_var, _, _ = paddle.static.auc(input=predict, label=label)
        cost = paddle.nn.functional.cross_entropy(
            input=predict, label=label, reduction='none', use_softmax=False
        )
        avg_cost = paddle.mean(x=cost)

        self.feeds = data
        self.train_file_path = ["fake1", "fake2"]
        self.avg_cost = avg_cost
        self.predict = predict

        return avg_cost

    def do_pyreader_training(self, fleet):
        """
        do training using dataset, using fetch handler to catch variable
        Args:
            fleet(Fleet api): the fleet object of Parameter Server, define distribute training role
        """

        exe = base.Executor(base.CPUPlace())

        exe.run(base.default_startup_program())
        fleet.init_worker()

        batch_size = 4

        train_reader = paddle.batch(fake_ctr_reader(), batch_size=batch_size)
        self.reader.decorate_sample_list_generator(train_reader)

        for epoch_id in range(1):
            self.reader.start()
            try:
                while True:
                    loss_val = exe.run(
                        program=base.default_main_program(),
                        fetch_list=[self.avg_cost.name],
                    )
                    loss_val = np.mean(loss_val)
                    print(f"TRAIN ---> pass: {epoch_id} loss: {loss_val}\n")
            except base.core.EOFException:
                self.reader.reset()

        model_dir = os.getenv("MODEL_DIR", None)
        if model_dir:
            fleet.save_inference_model(
                exe,
                model_dir,
                [feed.name for feed in self.feeds],
                self.avg_cost,
            )
            fleet.load_model(model_dir, mode=1)


if __name__ == "__main__":
    runtime_main(TestDistCTR2x2)