tutorials/mindalpha-tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "copyrighted-aluminum",
   "metadata": {
    "tags": []
   },
   "source": [
    "# MindAlpha Tutorial"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fixed-triangle",
   "metadata": {},
   "source": [
    "In this tutorial, we show how to use MindAlpha in the production environment."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ordered-option",
   "metadata": {
    "tags": []
   },
   "source": [
    "Before you proceed, please make sure you have uploaded the demo dataset to your own s3 bucket. See the **Prepare Data** section in [MindAlpha Getting Started](mindalpha-getting-started.ipynb) for instructions. In the rest of the article, we assume the demo dataset has been uploaded to ``s3://{YOUR_S3_BUCKET}/{YOUR_S3_PATH}/demo/data/``. You should replace ``{YOUR_S3_BUCKET}`` and ``{YOUR_S3_PATH}`` with actual values before executing code cells containing these placeholders."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "abstract-dealing",
   "metadata": {},
   "source": [
    "The ``schema`` directory contains configuration files for ``ma.EmbeddingSumConcat`` operators and must also be uploaded to s3. In the rest of the article, we assume the ``schema`` directory has been uploaded to ``s3://{YOUR_S3_BUCKET}/{YOUR_S3_PATH}/demo/schema/``."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "southwest-cambridge",
   "metadata": {},
   "source": [
    "If uploading hasn't been done, you can open a terminal by selecting the ``File`` -> ``New`` -> ``Terminal`` menu item and executing Bash commands similar to the following in it to upload these files to your own s3 bucket."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "spread-pharmaceutical",
   "metadata": {
    "tags": []
   },
   "source": [
    "```text\n",
    "aws s3 cp --recursive ${PWD}/data/ s3://{YOUR_S3_BUCKET}/{YOUR_S3_PATH}/demo/data/\n",
    "aws s3 cp --recursive ${PWD}/schema/ s3://{YOUR_S3_BUCKET}/{YOUR_S3_PATH}/demo/schema/\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "intermediate-treatment",
   "metadata": {},
   "outputs": [],
   "source": [
    "S3_ROOT_DIR = 's3://{YOUR_S3_BUCKET}/{YOUR_S3_PATH}/'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "lyric-safety",
   "metadata": {},
   "source": [
    "## Define the Model"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "professional-ordinary",
   "metadata": {},
   "source": [
    "Let's define our neural network model as the following ``DemoModule`` class. The is the same ``DemoModule`` class defined in [MindAlpha Getting Started](mindalpha-getting-started.ipynb)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "standing-visit",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import torch\n",
    "import mindalpha as ma\n",
    "\n",
    "class DemoModule(torch.nn.Module):\n",
    "    def __init__(self):\n",
    "        super().__init__()\n",
    "        self._embedding_size = 16\n",
    "        self._schema_dir = S3_ROOT_DIR + 'demo/schema/'\n",
    "        self._column_name_path = self._schema_dir + 'column_name_demo.txt'\n",
    "        self._combine_schema_path = self._schema_dir + 'combine_schema_demo.txt'\n",
    "        self._sparse = ma.EmbeddingSumConcat(self._embedding_size, self._column_name_path, self._combine_schema_path)\n",
    "        self._sparse.updater = ma.FTRLTensorUpdater()\n",
    "        self._sparse.initializer = ma.NormalTensorInitializer(var=0.01)\n",
    "        self._dense = torch.nn.Sequential(\n",
    "            ma.nn.Normalization(self._sparse.feature_count * self._embedding_size),\n",
    "            torch.nn.Linear(self._sparse.feature_count * self._embedding_size, 1024),\n",
    "            torch.nn.ReLU(),\n",
    "            torch.nn.Linear(1024, 512),\n",
    "            torch.nn.ReLU(),\n",
    "            torch.nn.Linear(512, 1),\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self._sparse(x)\n",
    "        x = self._dense(x)\n",
    "        return torch.sigmoid(x)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "collectible-manner",
   "metadata": {},
   "source": [
    "## Define the train function"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "polyphonic-peter",
   "metadata": {},
   "source": [
    "Now we define the ``train()`` function to wrap all the logic in one function to make it easier to develop the model in Jupyter locally and submit it to Airflow later. This basicly combines the training and evaluation steps into a single function."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "august-cyprus",
   "metadata": {},
   "source": [
    "Most of the code fragments have been shown in [MindAlpha Getting Started](mindalpha-getting-started.ipynb).\n",
    "\n",
    "``model_in_path`` specifies where to load a previously trained model from. With ``model_in_path`` and ``model_out_path``, we can train the model incrementally. If ``model_in_path`` is ``None``, the model is randomly initialized and trained from scratch.\n",
    "\n",
    "``model_export_path``, ``model_version`` and ``experiment_name`` are used for model exporting. An exported model can be loaded by MindAlpha Serving for online prediction. For the exported model to be found by MindAlpha Serving, we need to set ``consul_host``, ``consul_port`` and ``consul_endpoint_prefix`` and call the ``PyTorchModel.publish()`` method.\n",
    "\n",
    "``max_sparse_feature_age`` limits the existence of sparse features. If the embedding vector of a sparse feature is not updated for more than ``max_sparse_feature_age`` periods, it will finally be cleared out from the model. The value of ``max_sparse_feature_age`` should be adjusted to match model training frequency. For example, for daily training data, ``max_sparse_feature_age == 15`` means 15 days, whereas, for hourly training data, ``max_sparse_feature_age == 15 * 24`` means 15 days."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ea09823-6be9-4a6d-9088-2903d38160f4",
   "metadata": {},
   "source": [
    "We use Python's with statement to ensure the invocation of the ``stop()`` method of ``spark_session``."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "organized-remedy",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def train(local=True,\n",
    "          batch_size=100,\n",
    "          worker_count=1,\n",
    "          server_count=1,\n",
    "          worker_cpu=1,\n",
    "          server_cpu=1,\n",
    "          worker_memory='5G',\n",
    "          server_memory='5G',\n",
    "          coordinator_memory='5G',\n",
    "          module_class=None,\n",
    "          model_in_path=None,\n",
    "          model_out_path=None,\n",
    "          model_export_path=None,\n",
    "          model_version=None,\n",
    "          experiment_name=None,\n",
    "          input_label_column_index=0,\n",
    "          delimiter='\\t',\n",
    "          train_dataset_path=None,\n",
    "          test_dataset_path=None,\n",
    "          is_catchup=True,\n",
    "          consul_host=None,\n",
    "          consul_port=None,\n",
    "          consul_endpoint_prefix=None,\n",
    "          max_sparse_feature_age=15,\n",
    "          metric_update_interval=10,\n",
    "         ):\n",
    "    import pyspark\n",
    "    import mindalpha as ma\n",
    "    if module_class is None:\n",
    "        module_class = DemoModule\n",
    "    print('local: %s' % local)\n",
    "    print('batch_size: %d' % batch_size)\n",
    "    print('worker_count: %d' % worker_count)\n",
    "    print('server_count: %d' % server_count)\n",
    "    print('worker_cpu: %d' % worker_cpu)\n",
    "    print('server_cpu: %d' % server_cpu)\n",
    "    print('worker_memory: %s' % worker_memory)\n",
    "    print('server_memory: %s' % server_memory)\n",
    "    print('coordinator_memory: %s' % coordinator_memory)\n",
    "    print('module_class: %s' % module_class)\n",
    "    print('model_in_path: %s' % model_in_path)\n",
    "    print('model_out_path: %s' % model_out_path)\n",
    "    print('model_export_path: %s' % model_export_path)\n",
    "    print('model_version: %s' % model_version)\n",
    "    print('experiment_name: %s' % experiment_name)\n",
    "    print('input_label_column_index: %d' % input_label_column_index)\n",
    "    print('delimiter: %r' % delimiter)\n",
    "    print('train_dataset_path: %s' % train_dataset_path)\n",
    "    print('test_dataset_path: %s' % test_dataset_path)\n",
    "    print('is_catchup: %s' % is_catchup)\n",
    "    print('consul_host: %s' % consul_host)\n",
    "    print('consul_port: %s' % consul_port)\n",
    "    print('consul_endpoint_prefix: %s' % consul_endpoint_prefix)\n",
    "    print('max_sparse_feature_age: %d' % max_sparse_feature_age)\n",
    "    print('metric_update_interval: %d' % metric_update_interval)\n",
    "    module = module_class()\n",
    "    estimator = ma.PyTorchEstimator(module=module,\n",
    "                                    worker_count=worker_count,\n",
    "                                    server_count=server_count,\n",
    "                                    model_in_path=model_in_path,\n",
    "                                    model_out_path=model_out_path,\n",
    "                                    model_export_path=model_export_path,\n",
    "                                    model_version=model_version,\n",
    "                                    experiment_name=experiment_name,\n",
    "                                    input_label_column_index=input_label_column_index,\n",
    "                                    consul_host=consul_host,\n",
    "                                    consul_port=consul_port,\n",
    "                                    consul_endpoint_prefix=consul_endpoint_prefix,\n",
    "                                    max_sparse_feature_age=max_sparse_feature_age,\n",
    "                                    metric_update_interval=metric_update_interval,\n",
    "                                   )\n",
    "    spark_session = ma.spark.get_session(local=local,\n",
    "                                         batch_size=batch_size,\n",
    "                                         worker_count=estimator.worker_count,\n",
    "                                         server_count=estimator.server_count,\n",
    "                                         worker_cpu=worker_cpu,\n",
    "                                         server_cpu=server_cpu,\n",
    "                                         worker_memory=worker_memory,\n",
    "                                         server_memory=server_memory,\n",
    "                                         coordinator_memory=coordinator_memory,\n",
    "                                        )\n",
    "    with spark_session:\n",
    "        train_dataset = ma.input.read_s3_csv(spark_session, train_dataset_path, delimiter=delimiter,\n",
    "                                             shuffle=True, num_workers=estimator.worker_count)\n",
    "        model = estimator.fit(train_dataset)\n",
    "        if test_dataset_path is not None:\n",
    "            test_dataset = ma.input.read_s3_csv(spark_session, test_dataset_path, delimiter=delimiter)\n",
    "            result = model.transform(test_dataset)\n",
    "            evaluator = pyspark.ml.evaluation.BinaryClassificationEvaluator()\n",
    "            test_auc = evaluator.evaluate(result)\n",
    "            print('test_auc: %g' % test_auc)\n",
    "        if not is_catchup and model.consul_endpoint_prefix is not None:\n",
    "            model.publish()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "spoken-budapest",
   "metadata": {},
   "source": [
    "We can use example paths to call the ``train()`` function to test our model. Later, if you changed the class definition of the model, you can call the ``train()`` function again to test, which makes it convenient to develop the model in Jupyter interactively."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "outer-plenty",
   "metadata": {},
   "outputs": [],
   "source": [
    "model_out_path = S3_ROOT_DIR + 'demo/output/dev/model_out/'\n",
    "train_dataset_path = S3_ROOT_DIR + 'demo/data/train/day_0_0.001_train.csv'\n",
    "test_dataset_path = S3_ROOT_DIR + 'demo/data/test/day_0_0.001_test.csv'\n",
    "train(model_out_path=model_out_path,\n",
    "      train_dataset_path=train_dataset_path,\n",
    "      test_dataset_path=test_dataset_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "contemporary-mount",
   "metadata": {},
   "source": [
    "## Schedule model training"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "broke-registrar",
   "metadata": {},
   "source": [
    "To schedule model training, let's define the following variables.\n",
    "\n",
    "The first group of variables identify our experiment. ``business_name`` is the name of the machine learning application in your organization. ``experiment_name`` specifies a name for the iteration of the model. ``job_name`` specifies a name for the machine learning task to distinguish it from other tasks in the Airflow DAG (such as data preprocessing tasks).\n",
    "\n",
    "The second group of variables are related to Airflow. ``owner`` specifies Airflow DAG owner. ``schedule_interval`` specifies the scheduling frequency of model training. ``backfill_start_date`` and ``backfill_end_date`` specify the start and end date of the model backfill process, which are Python ``datetime.datetime`` in UTC timezone actually, but we name them \"dates\" to match the terms of Airflow.\n",
    "\n",
    "The demo dataset contains data of 24 days. The rest of the document uses daily data to simulate 5 minutes ``schedule_interval`` model training so that the running of the demo can finish quickly. We compute ``backfill_start_date`` and ``backfill_end_date`` based on the current time so that the backfill process starts from 15 minutes (3 schedule intervals) ago and ends at 15 minutes (3 schedule intervals also) later from now. ``online_start_date`` is 20 minutes later from now. In real applications, ``backfill_start_date`` and ``backfill_end_date`` may be specified directly instead of computed, and you may would like to use an ``'@hourly'`` or ``'@daily'`` ``schedule_interval``.\n",
    "\n",
    "**Note**\n",
    "\n",
    "1. It's important to specify a **future** time for ``backfill_end_date``, otherwise there may be gaps between the end of the backfill process and the start of the first online task, and the ``model_in_path`` of the first online task will be incorrect.\n",
    "2. ``backfill_start_date``, ``backfill_end_date`` and ``online_start_date`` must be timezone-aware ``datetime.datetime``s. Non-UTC timezone can be used, but Airflow always passes timezone-aware ``datetime.datetime`` in UTC timezone for the ``execution_date`` parameter of the ``execute()`` function mentioned later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cloudy-swedish",
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "business_name = 'jupyter_doc'\n",
    "experiment_name = 'ma_tutorial'\n",
    "job_name = 'train'\n",
    "owner = 'admin'\n",
    "schedule_interval = datetime.timedelta(minutes=5)\n",
    "utc_now = datetime.datetime.utcnow().timestamp()\n",
    "backfill_timestamp = int(utc_now / schedule_interval.total_seconds()) * schedule_interval.total_seconds()\n",
    "backfill_start_date = datetime.datetime.fromtimestamp(backfill_timestamp, tz=datetime.timezone.utc) - schedule_interval * 3\n",
    "backfill_end_date = backfill_start_date + schedule_interval * 5\n",
    "online_start_date = backfill_end_date + schedule_interval\n",
    "print(f'backfill_start_date: {backfill_start_date}')\n",
    "print(f'backfill_end_date  : {backfill_end_date}')\n",
    "print(f'online_start_date  : {online_start_date}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "hidden-leave",
   "metadata": {},
   "source": [
    "For the trained model to be loaded by MindAlpha Serving for online prediction, the following variables should be set properly, see the documentation of MindAlpha Serving fore more information."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "stable-china",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "consul_host = None\n",
    "consul_port = None\n",
    "consul_endpoint_prefix = None\n",
    "\n",
    "# consul_host = 'consul-host.example.com'\n",
    "# consul_port = 8500\n",
    "# consul_endpoint_prefix = 'demo/mindalpha-models'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "union-dressing",
   "metadata": {},
   "source": [
    "Next, we define the ``execute()`` function to be invoked by Airflow every ``schedule_interval``.\n",
    "\n",
    "``execute()`` uses ``execution_date`` to compute the arguments of ``train()`` and then calls ``train()`` with the computed arguments to train a model for ``execution_date``. Be careful to set ``model_in_path`` to ``None`` for the first model, otherwise ``train()`` will fail.\n",
    "\n",
    "The demo code here computes ``data_part_index`` at first and then uses it to compute dateset paths. In real applications, we can usually compute dataset paths directly based on ``execution_date``.\n",
    "\n",
    "The ``is_catchup`` parameter will be ``True`` for backfill invocation and ``False`` for online invocation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "adequate-alaska",
   "metadata": {},
   "outputs": [],
   "source": [
    "def execute(execution_date, is_catchup, **kwargs):\n",
    "    import datetime\n",
    "    print('Train model for %s' % execution_date.isoformat())\n",
    "    train_dataset_dir = S3_ROOT_DIR + 'demo/data/train/'\n",
    "    test_dataset_dir = S3_ROOT_DIR + 'demo/data/test/'\n",
    "    output_dir = S3_ROOT_DIR + 'demo/output/model_out/'\n",
    "    export_dir = S3_ROOT_DIR + 'demo/output/model_export/'\n",
    "    data_part_index = int(round((execution_date - backfill_start_date).total_seconds() / schedule_interval.total_seconds()))\n",
    "    model_version_format = '%Y%m%d%H%M'\n",
    "    model_version = execution_date.strftime(model_version_format)\n",
    "    train_dataset_path = train_dataset_dir + 'day_%d_0.001_train.csv' % data_part_index\n",
    "    test_dataset_path = test_dataset_dir + 'day_%d_0.001_test.csv' % data_part_index\n",
    "    model_in_path = None\n",
    "    if data_part_index > 0:\n",
    "        previous_execution_date = execution_date - schedule_interval\n",
    "        previous_model_version = previous_execution_date.strftime(model_version_format)\n",
    "        model_in_path = output_dir + '%s/' % previous_model_version\n",
    "    model_out_path = output_dir + '%s/' % model_version\n",
    "    model_export_path = export_dir + '%s/' % model_version\n",
    "    train(local=False,\n",
    "          batch_size=100,\n",
    "          worker_count=10,\n",
    "          server_count=10,\n",
    "          module_class=DemoModule,\n",
    "          model_in_path=model_in_path,\n",
    "          model_out_path=model_out_path,\n",
    "          model_export_path=model_export_path,\n",
    "          model_version=model_version,\n",
    "          experiment_name=experiment_name,\n",
    "          train_dataset_path=train_dataset_path,\n",
    "          test_dataset_path=test_dataset_path,\n",
    "          is_catchup=is_catchup,\n",
    "          consul_host=consul_host,\n",
    "          consul_port=consul_port,\n",
    "          consul_endpoint_prefix=consul_endpoint_prefix,\n",
    "         )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dutch-model",
   "metadata": {},
   "source": [
    "Now, we can create an ``ma.experiment.Experiment`` object and call its ``submit_backfill()`` method to let Airflow schedule training models for history data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "legitimate-affiliate",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "experiment = ma.experiment.Experiment(business_name=business_name,\n",
    "                                      experiment_name=experiment_name,\n",
    "                                      job_name=job_name,\n",
    "                                      owner=owner,\n",
    "                                      schedule_interval=schedule_interval,\n",
    "                                      func=execute,\n",
    "                                      start_date=backfill_start_date,\n",
    "                                      end_date=backfill_end_date,\n",
    "                                      extra_dag_conf={'depends_on_past': True})\n",
    "experiment.submit_backfill()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "exciting-stephen",
   "metadata": {},
   "source": [
    "To schedule training models for real-time data, we can create an ``ma.experiment.Experiment`` object and call its ``submit_backfill()`` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "suburban-morning",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "experiment = ma.experiment.Experiment(business_name=business_name,\n",
    "                                      experiment_name=experiment_name,\n",
    "                                      job_name=job_name,\n",
    "                                      owner=owner,\n",
    "                                      schedule_interval=schedule_interval,\n",
    "                                      func=execute,\n",
    "                                      start_date=online_start_date,\n",
    "                                      extra_dag_conf={'depends_on_past': True})\n",
    "experiment.submit_online()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "south-passing",
   "metadata": {},
   "source": [
    "## Summary"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "competitive-screening",
   "metadata": {},
   "source": [
    "We illustrated how to use MindAlpha in the production environment. We defined the ``train()`` function which can be used to develop models in Jupyter notebook interactively and called by ``execute()`` to train models incrementally."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}