examples/dag-example/count_words.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DAG Example: Map-Reduce word count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This example DAG consists of counting the words of some books available for free at the [Gutenberg Project website](https://www.gutenberg.org). To do so, a Map task counts the words in parallel by splitting the texts in chunks. Then, for every book all the words counted are merged and reduced in a single function."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Prerequisites to run this example:\n",
    "\n",
    "In order to run this example, you will need:\n",
    "\n",
    "- An IBM Cloud Functions account. You can register for free and access the IBM Cloud Functions free tier [here](https://www.ibm.com/cloud/free).\n",
    "\n",
    "- The [IBM Cloud CLI](https://www.ibm.com/cloud/cli) with the IBM Cloud Functions plugin installed and set up. Make sure you are logged in and the correct region and IBM Cloud Functions namespace is selected.\n",
    "\n",
    "- A [DockerHub](https://hub.docker.com/) account and [Docker CE](https://docs.docker.com/get-docker/) installed in your system.\n",
    "\n",
    "- A Triggerflow installation up and running. Please refer to the installation guide to properly install a Triggerflow instance."
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "#### Setup\n",
    "\n",
    "In this section we will build the Docker runtime image and create the IBM Cloud functions:\n",
    "\n",
    "1. Build the Triggerflow runtime for IBM Cloud Functions: Go to [IBM Cloud Functions runtime directory](../../runtime/ibm_cloud_functions) and build the image depending on you Python version. For example, for Python 3.8:\n",
    "\n",
    "```bash\n",
    "$ docker build . -f Dockerfile_v38 -t ${YOUR_DOCKER_USERNAME}/triggerflow_runtime:v3.8-0\n",
    "```\n",
    "\n",
    "```bash\n",
    "$ docker push ${YOUR_DOCKER_USERNAME}/triggerflow_runtime:v3.8-0\n",
    "```\n",
    "\n",
    "Replace `${YOUR_DOCKER_USERNAME}` with your DockerHub user name.\n",
    "\n",
    "2. Create the IBM Cloud Function actions used in this workflow:\n",
    "\n",
    "```bash\n",
    "$ ibmcloud cloud-functions package create triggerflow-count-words\n",
    "```\n",
    "\n",
    "```bash\n",
    "$ ibmcloud cloud-functions action create triggerflow-count-words/count-words --docker ${YOUR_DOCKER_USERNAME}/triggerflow_runtime:v3.8-0 count_words.py\n",
    "```\n",
    "\n",
    "```bash\n",
    "$ ibmcloud cloud-functions action create triggerflow-count-words/merge-dicts --docker ${YOUR_DOCKER_USERNAME}/triggerflow_runtime:v3.8-0 merge_dicts.py\n",
    "```"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   }
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## DAG Definition\n",
    "\n",
    "The cell below descirbes the DAG tasks and dependencies. Similarly to Airflow's DAG definition, a Triggerflow DAG is composed of multiple operators that describe how or where a task is executed. In this case, we are executing serverless functions on IBM Cloud. The bitwise shift operator is used to set the task order of execution. For example, here the `count_*` tasks will be executed before the `merge_*` tasks. The `merge_*` are getting a JSONPath as value for the parameter `dicts`. This JSONPath is used to pass the data from a task to a following one. The JSONPath is parsed and evaluated in runtime when the task is being prepared to be run. In this case, with `$.[*]`, we are getting all the elements from the array of results that the `count_*` tasks generate."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from triggerflow.dags import DAG\n",
    "from triggerflow.dags.operators import IBMCloudFunctionsCallAsyncOperator, IBMCloudFunctionsMapOperator\n",
    "\n",
    "urls = ['https://www.gutenberg.org/files/1342/1342-0.txt',\n",
    "        'https://www.gutenberg.org/files/11/11-0.txt',\n",
    "        'https://www.gutenberg.org/files/1661/1661-0.txt']\n",
    "\n",
    "# urls = ['http://www.gutenberg.org/files/2701/2701-0.txt']\n",
    "\n",
    "dag = DAG('count-words')\n",
    "parallel = 8\n",
    "\n",
    "for i, url in enumerate(urls):\n",
    "    count = IBMCloudFunctionsMapOperator(\n",
    "        task_id='count_{}'.format(i),\n",
    "        function_name='count-words',\n",
    "        function_package='triggerflow-count-words',\n",
    "        invoke_kwargs={'url': url,\n",
    "                       'parallel': parallel},\n",
    "        iter_data=('func_id', [par for par in range(parallel)]),\n",
    "        dag=dag)\n",
    "    \n",
    "    merge = IBMCloudFunctionsCallAsyncOperator(\n",
    "        task_id='merge_{}'.format(i),\n",
    "        function_name='merge-dicts',\n",
    "        function_package='triggerflow-count-words',\n",
    "        invoke_kwargs={'dicts': '$.[*]'.format(i)},\n",
    "        dag=dag)\n",
    "    \n",
    "    count >> merge"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can visualize the DAG tasks and order of execution using `dag.show()`, which uses Grpahviz as backend to plot the graph image for now."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dag.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Triggerlfow allows to save DAGs in a local folder of this system. By doing so, we could trigger a DAG execution without having to redeclare the whole DAG object as before. Triggerflow stores its DAGs in JSON format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dag.save()\n",
    "\n",
    "some_new_dag_object = DAG('count-words')\n",
    "some_new_dag_object.load()\n",
    "some_new_dag_object.tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## DAG execution\n",
    "\n",
    "We can trigger a DAG exection using the `dag.run()` method. This will create a DAGRun object, with a unique ID of execution. The DAGRun object, when it is instantiated, it registers all the triggers neccessary to orchestarte the DAG, and publishes the initial CloudEvent to the event source, so that the DAG starts executing right away."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "dagrun = dag.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `dagrun.result()` gets the result of the final task. The result are the `data` values from the events that activated the final join trigger. A task_id can be passed as parameter to retrieve the results from any other task in the DAG. However, this method call does not block until the DAG execution is completed, so it might raise an exception if the results aren't yet available."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "result = dagrun.result()\n",
    "print(result[1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "DAGRuns are also stored in the local system's Triggerflow cache. This is useful for long running DAG executions, because the state of the workflow can be retreived without having to maintain instantiated the DAGRun object. To load a run, the ID is required."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from triggerflow.dags import DAGRun\n",
    "\n",
    "same_run = DAGRun.load_run('count-words-f123c4dc5bd2')\n",
    "res = same_run.result()\n",
    "print(res)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "name": "python3",
   "language": "python",
   "display_name": "Python 3"
  },
  "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.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}