Developing_With_Apache_Beam_Cleaned.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "R_vFbXN2T7oo"
   },
   "source": [
    "##### Copyright &copy; 2020 The TensorFlow Authors."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "NhL09apzT-W_"
   },
   "outputs": [],
   "source": [
    "#@title Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "# https://urldefense.com/v3/__https://www.apache.org/licenses/LICENSE-2.0*5Cn__;JQ!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4Rxhel3RfE$ #\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "23R0Z9RojXYW"
   },
   "source": [
    "# TFX – Developing With Apache Beam\n",
    "\n",
    "[Apache Beam](https://urldefense.com/v3/__https://www.tensorflow.org/tfx/guide/beam__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxlMaOx6c$ ) is an open source, unified model for defining both batch and streaming data-parallel processing pipelines. TFX uses Apache Beam to implement data-parallel pipelines. The pipeline is then executed by one of Beam's supported distributed processing back-ends or \"runners\", which include [Dataflow](https://urldefense.com/v3/__https://cloud.google.com/dataflow__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4Rx_xDiz6Q$ ), [Apache Flink](https://urldefense.com/v3/__https://flink.apache.org/__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxUIpCch4$ ), [Apache Spark](https://urldefense.com/v3/__https://spark.apache.org/__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxVA2PkcU$ ), [Apache Samza](https://urldefense.com/v3/__http://samza.apache.org/__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4Rx5goppac$ ), and others.\n",
    "\n",
    "TFX is designed to be scalable to very large datasets which require substantial resources.  Distributed pipeline frameworks such as Apache Beam offer the ability to distribute processing across compute clusters and apply the resources required.  Many of the standard TFX components use Apache Beam, and custom components that you may write may also benefit from using Apache Beam for distibuted processing.\n",
    "\n",
    "This notebook introduces the concepts and code patterns for developing with the Apache Beam Python API. This is especially important for developers who will be creating custom TFX components which may require substantial computing resources, and will benefit from distributed processing using beam."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "EnOzbFhBKqwf"
   },
   "source": [
    "### [DirectRunner](https://beam.apache.org/documentation/runners/direct/)\n",
    "\n",
    "In this notebook, to avoid the requirement for a distributed processing cluster we will instead use Beam's [`DirectRunner`](https://urldefense.com/v3/__https://beam.apache.org/documentation/runners/direct/__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxE4HG0-g$ ).  The `DirectRunner` is also very convenient when developing Beam pipelines. The `DirectRunner` executes pipelines on your machine and is designed to validate that pipelines adhere to the Beam programming model as closely as possible. Instead of focusing on efficient pipeline execution, the `DirectRunner` performs additional checks to ensure that users do not rely on semantics that are not guaranteed by the model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "2GivNBNYjb3b"
   },
   "source": [
    "## Setup\n",
    "First, we install the necessary packages, download data, import modules and set up paths.\n",
    "\n",
    "### Install Apache Beam"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T14:54:15.413669Z",
     "start_time": "2021-04-13T14:54:15.289007Z"
    },
    "id": "7zhcJBLoMXQE"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "zsh:1: no matches found: apache-beam[interactive]\r\n"
     ]
    }
   ],
   "source": [
    "!pip install -q -U \\\n",
    "  apache-beam \\\n",
    "  apache-beam[interactive]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "N-ePgV0Lj68Q"
   },
   "source": [
    "### Import packages\n",
    "We import necessary packages, including Beam."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T14:55:15.886020Z",
     "start_time": "2021-04-13T14:55:11.728879Z"
    }
   },
   "outputs": [],
   "source": [
    "!pip install -q -U graphviz"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T14:55:18.248778Z",
     "start_time": "2021-04-13T14:55:17.870588Z"
    },
    "id": "YIqpWK9efviJ"
   },
   "outputs": [],
   "source": [
    "import apache_beam as beam\n",
    "from apache_beam import pvalue\n",
    "from apache_beam.runners.interactive.display import pipeline_graph\n",
    "import graphviz"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "p4XfH7dyRZtd"
   },
   "source": [
    "Check the version."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T14:55:21.491634Z",
     "start_time": "2021-04-13T14:55:21.489243Z"
    },
    "id": "0uuFfIiHviyW"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Beam version: 2.28.0\n"
     ]
    }
   ],
   "source": [
    "print('Beam version: {}'.format(beam.__version__))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "nCb-bS_eMkMf"
   },
   "source": [
    "## Beam Python Syntax\n",
    "\n",
    "Beam uses a [special Python syntax](https://urldefense.com/v3/__https://beam.apache.org/documentation/programming-guide/*applying-transforms__;Iw!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4Rxw19zxq8$ ) to define and invoke transforms.  For example, in this line:\n",
    "\n",
    ">`result = pass_this | 'name this step' >> to_this_call`\n",
    "\n",
    "The method `to_this_call` is being invoked and passed the object called `pass_this`, and this operation will be referred to as [name this step](https://urldefense.com/v3/__https://stackoverflow.com/questions/50519662/what-does-the-redirection-mean-in-apache-beam-python__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxMFC5UkE$ ) in a stack trace or pipeline diagram.  The result of the call to `to_this_call` is returned in `result`.  You will often see stages of a pipeline chained together like this:\n",
    "\n",
    ">`result = apache_beam.Pipeline() | 'first step' >> do_this_first() | 'second step' >> do_this_last()`\n",
    "\n",
    "and since that started with a new pipeline, you can continue like this:\n",
    "\n",
    ">`next_result = result | 'doing more stuff' >> another_function()`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "qAika7-6gLvI"
   },
   "source": [
    "## Create a Beam Pipeline\n",
    "\n",
    "Create a pipeline, including a simple [`PCollection`](https://urldefense.com/v3/__https://beam.apache.org/releases/javadoc/2.1.0/org/apache/beam/sdk/values/PCollection.html__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4Rxgmi9VMc$ ) and a [`ParDo()`](https://urldefense.com/v3/__https://beam.apache.org/releases/javadoc/2.0.0/org/apache/beam/sdk/transforms/ParDo.html__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxfbsGH5U$ ) transform.\n",
    "\n",
    "* A `PCollection<T>` is an **immutable collection** of values of type `T`. A `PCollection` can contain either a bounded or unbounded number of elements. Bounded and unbounded `PCollections` are produced as the output of `PTransforms` (including root `PTransforms` like `Read` and `Create`), and can be passed as the inputs of other `PTransforms`.\n",
    "* `ParDo` is the core **element-wise** transform in Apache Beam, invoking a user-specified function on each of the elements of the input `PCollection` to produce zero or more output elements, all of which are collected into the output `PCollection`.\n",
    "\n",
    "This can be done in different ways, depending on coding style preferences. First, use the `.run()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T14:55:27.406501Z",
     "start_time": "2021-04-13T14:55:26.964368Z"
    },
    "id": "X_XV_cm2_E1x"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:apache_beam.runners.interactive.interactive_environment:Dependencies required for Interactive Beam PCollection visualization are not available, please use: `pip install apache-beam[interactive]` to install necessary dependencies to enable all data visualization features.\n"
     ]
    },
    {
     "data": {
      "application/javascript": [
       "\n",
       "        if (typeof window.interactive_beam_jquery == 'undefined') {\n",
       "          var jqueryScript = document.createElement('script');\n",
       "          jqueryScript.src = 'https://code.jquery.com/jquery-3.4.1.slim.min.js';\n",
       "          jqueryScript.type = 'text/javascript';\n",
       "          jqueryScript.onload = function() {\n",
       "            var datatableScript = document.createElement('script');\n",
       "            datatableScript.src = 'https://cdn.datatables.net/1.10.20/js/jquery.dataTables.min.js';\n",
       "            datatableScript.type = 'text/javascript';\n",
       "            datatableScript.onload = function() {\n",
       "              window.interactive_beam_jquery = jQuery.noConflict(true);\n",
       "              window.interactive_beam_jquery(document).ready(function($){\n",
       "                \n",
       "              });\n",
       "            }\n",
       "            document.head.appendChild(datatableScript);\n",
       "          };\n",
       "          document.head.appendChild(jqueryScript);\n",
       "        } else {\n",
       "          window.interactive_beam_jquery(document).ready(function($){\n",
       "            \n",
       "          });\n",
       "        }"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hello\n",
      "World\n",
      "!!!\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'DONE'"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_pipeline = beam.Pipeline()\n",
    "\n",
    "lines = (first_pipeline\n",
    "         | \"Create\" >> beam.Create([\"Hello\", \"World\", \"!!!\"]) # PCollection\n",
    "         | \"Print\" >> beam.ParDo(print)) # ParDo transform\n",
    "\n",
    "result = first_pipeline.run()\n",
    "result.state"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "qx1xFZ-BYccz"
   },
   "source": [
    "Display the structure of this pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T14:55:37.409017Z",
     "start_time": "2021-04-13T14:55:37.208478Z"
    },
    "id": "bdnEk2p3Jnfs",
    "scrolled": true
   },
   "outputs": [
    {
     "ename": "ExecutableNotFound",
     "evalue": "failed to execute ['dot', '-Kdot', '-Tsvg'], make sure the Graphviz executables are on your systems' PATH",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mFileNotFoundError\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/site-packages/graphviz/backend.py\u001b[0m in \u001b[0;36mrun\u001b[0;34m(cmd, input, capture_output, check, encoding, quiet, **kwargs)\u001b[0m\n\u001b[1;32m    163\u001b[0m     \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 164\u001b[0;31m         \u001b[0mproc\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0msubprocess\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPopen\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcmd\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstartupinfo\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mget_startupinfo\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    165\u001b[0m     \u001b[0;32mexcept\u001b[0m \u001b[0mOSError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/subprocess.py\u001b[0m in \u001b[0;36m__init__\u001b[0;34m(self, args, bufsize, executable, stdin, stdout, stderr, preexec_fn, close_fds, shell, cwd, env, universal_newlines, startupinfo, creationflags, restore_signals, start_new_session, pass_fds, encoding, errors, text)\u001b[0m\n\u001b[1;32m    853\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 854\u001b[0;31m             self._execute_child(args, executable, preexec_fn, close_fds,\n\u001b[0m\u001b[1;32m    855\u001b[0m                                 \u001b[0mpass_fds\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcwd\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0menv\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/subprocess.py\u001b[0m in \u001b[0;36m_execute_child\u001b[0;34m(self, args, executable, preexec_fn, close_fds, pass_fds, cwd, env, startupinfo, creationflags, shell, p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite, restore_signals, start_new_session)\u001b[0m\n\u001b[1;32m   1701\u001b[0m                         \u001b[0merr_msg\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mos\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstrerror\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0merrno_num\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1702\u001b[0;31m                     \u001b[0;32mraise\u001b[0m \u001b[0mchild_exception_type\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0merrno_num\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0merr_msg\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0merr_filename\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1703\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mchild_exception_type\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0merr_msg\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mFileNotFoundError\u001b[0m: [Errno 2] No such file or directory: 'dot'",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[0;31mExecutableNotFound\u001b[0m                        Traceback (most recent call last)",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/site-packages/IPython/core/formatters.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, obj)\u001b[0m\n\u001b[1;32m    343\u001b[0m             \u001b[0mmethod\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mget_real_method\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobj\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprint_method\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    344\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mmethod\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 345\u001b[0;31m                 \u001b[0;32mreturn\u001b[0m \u001b[0mmethod\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    346\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    347\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/site-packages/graphviz/files.py\u001b[0m in \u001b[0;36m_repr_svg_\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    142\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    143\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_repr_svg_\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 144\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mpipe\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'svg'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdecode\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_encoding\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    145\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    146\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mpipe\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mformat\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mrenderer\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mformatter\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mquiet\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mFalse\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/site-packages/graphviz/files.py\u001b[0m in \u001b[0;36mpipe\u001b[0;34m(self, format, renderer, formatter, quiet)\u001b[0m\n\u001b[1;32m    167\u001b[0m         \u001b[0mdata\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mtext_type\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msource\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mencode\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_encoding\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    168\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 169\u001b[0;31m         out = backend.pipe(self._engine, format, data,\n\u001b[0m\u001b[1;32m    170\u001b[0m                            \u001b[0mrenderer\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mrenderer\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mformatter\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mformatter\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    171\u001b[0m                            quiet=quiet)\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/site-packages/graphviz/backend.py\u001b[0m in \u001b[0;36mpipe\u001b[0;34m(engine, format, data, renderer, formatter, quiet)\u001b[0m\n\u001b[1;32m    246\u001b[0m     \"\"\"\n\u001b[1;32m    247\u001b[0m     \u001b[0mcmd\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0m_\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mcommand\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mengine\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mformat\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mrenderer\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mformatter\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 248\u001b[0;31m     \u001b[0mout\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0m_\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mrun\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcmd\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0minput\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mdata\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcapture_output\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcheck\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mquiet\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mquiet\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    249\u001b[0m     \u001b[0;32mreturn\u001b[0m \u001b[0mout\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    250\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/site-packages/graphviz/backend.py\u001b[0m in \u001b[0;36mrun\u001b[0;34m(cmd, input, capture_output, check, encoding, quiet, **kwargs)\u001b[0m\n\u001b[1;32m    165\u001b[0m     \u001b[0;32mexcept\u001b[0m \u001b[0mOSError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    166\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0merrno\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0merrno\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mENOENT\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 167\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mExecutableNotFound\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcmd\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    168\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    169\u001b[0m             \u001b[0;32mraise\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mExecutableNotFound\u001b[0m: failed to execute ['dot', '-Kdot', '-Tsvg'], make sure the Graphviz executables are on your systems' PATH"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<graphviz.files.Source at 0x7f9ea1e67a00>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def display_pipeline(pipeline):\n",
    "    graph = pipeline_graph.PipelineGraph(pipeline)\n",
    "    return graphviz.Source(graph.get_dot())\n",
    "\n",
    "display_pipeline(first_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "JSaFbVyIYNIA"
   },
   "source": [
    "Using a different coding style, invoke run inside a `with` block."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "P4O8EAYNULbF"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as with_pipeline:\n",
    "  lines = (with_pipeline\n",
    "         | \"Create\" >> beam.Create([\"Hello\", \"World\", \"!!!\"])\n",
    "         | \"Print\" >> beam.ParDo(print))\n",
    "\n",
    "display_pipeline(with_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "yLJf3Zr5STWR"
   },
   "source": [
    "Notice that both pipelines and their outputs are the same."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "bMYiWcBtIBF7"
   },
   "source": [
    "### Now You Try\n",
    "\n",
    "This is a good point to stop reading and try writing code yourself, as a way to help you learn.\n",
    "\n",
    "**Exercise 1 — Creating and Running Your Beam Pipeline**\n",
    "\n",
    "1. Build a Beam pipeline that creates a PCollection containing integers 0 to 10 and prints them.\n",
    "2. Add a step in the pipeline to square each item.\n",
    "3. Display the pipeline.\n",
    "\n",
    "*Warning*: the `ParDo()` method must either return `None` or a list."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "0aPlwJjQKmo6"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "8f_Nf9XkH-vn"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "W46JR_uDH-lz"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "4f8NJsj-K1hG"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Ai6SmbkuK1SC"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "J-b7ZdFYLoH2"
   },
   "source": [
    "![Solution](https://urldefense.com/v3/__https://www.tensorflow.org/site-assets/images/marketing/learn/tfx-hero.svg__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxzvhZF7w$ )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Vbw50WdgMsLU"
   },
   "source": [
    "**Solution**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "PSUBzUGqfX4K"
   },
   "outputs": [],
   "source": [
    "def passthrough(label, x): # Utility for printing and returning the element\n",
    "  print(label, x)\n",
    "  return x\n",
    "\n",
    "# This includes printing stages for illustration.\n",
    "with beam.Pipeline() as with_pipeline:\n",
    "  lines = (with_pipeline\n",
    "         | \"Create\" >> beam.Create(range(10 + 1))\n",
    "         | 'Print0' >> beam.Map(lambda x: passthrough('x =', x))\n",
    "         | \"Square\" >> beam.ParDo(lambda x: [x ** 2])\n",
    "         | \"Print1\" >> beam.Map(lambda x: passthrough('x^2 =', x)))\n",
    "\n",
    "display_pipeline(with_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "C6lzLoIQz1Oe"
   },
   "source": [
    "# Core Transforms\n",
    "---\n",
    "Beam has a set of core transforms on data that is contained in `PCollections`.  In the cells that follow, explore several core transforms and observe the results in order to develop some understanding and intuition for what each transform does.\n",
    "\n",
    "## [Map](https://beam.apache.org/documentation/transforms/python/elementwise/map/)\n",
    "The `Map` transform applies a simple 1-to-1 mapping function over each element in the collection.  `Map` accepts a function that returns a single element for every input element in the `PCollection`.  You can pass functions with multiple arguments to `Map`. They are passed as additional positional arguments or keyword arguments to the function.\n",
    "\n",
    "First, compare the results of a `ParDo` transform and a `Map` transform for a multiply operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "bkzO7gANza49"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | \"Create\" >> beam.Create([1, 2, 3])\n",
    "           | 'Print0' >> beam.Map(lambda x: passthrough('x =', x))\n",
    "           | \"Multiply\" >> beam.ParDo(lambda number: [number * 2]) # ParDo with integers\n",
    "           | 'Print1' >> beam.Map(lambda x: passthrough('x * 2 =', x)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "dZWhKgqn02Ow"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "         | \"Create\" >> beam.Create([1, 2, 3])\n",
    "         | 'Print0' >> beam.Map(lambda x: passthrough('x =', x))\n",
    "         | \"Multiply\" >> beam.Map(lambda number: number * 2) # Map with integers\n",
    "         | 'Print1' >> beam.Map(lambda x: passthrough('x * 2 =', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "gh1R4stdUglj"
   },
   "source": [
    "Notice that the results are the same.  Now try it with a split operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "7bsEdlKj0Wce"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "         | \"Create\" >> beam.Create([\"Hello Beam\", \"This is cool\"])\n",
    "         | 'Print0' >> beam.Map(lambda x: passthrough('Sentence:', x))\n",
    "         | \"Split\" >> beam.ParDo(lambda sentence: sentence.split()) # ParDo with strings\n",
    "         | 'Print1' >> beam.Map(lambda x: passthrough('Word:', x)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "GY490-duaQFh"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "         | \"Create\" >> beam.Create([\"Hello Beam\", \"This is cool\"])\n",
    "         | 'Print0' >> beam.Map(lambda x: passthrough('Sentence:', x))\n",
    "         | \"Split\" >> beam.Map(lambda sentence: sentence.split()) # Map with strings\n",
    "         | 'Print1' >> beam.Map(lambda x: passthrough('Words:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "USH3bn-VcHsl"
   },
   "source": [
    "Notice that `ParDo` returned individual elements, while `Map` returned lists."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "wtcELhYDUsY0"
   },
   "source": [
    "## DoFn and [FlatMap](https://beam.apache.org/documentation/transforms/python/elementwise/flatmap/)\n",
    "Now try working with a `DoFn` in a `ParDo`, and compare that to using `FlatMap`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "t0-gNKm70oJD"
   },
   "outputs": [],
   "source": [
    "class BreakIntoWordsDoFn(beam.DoFn):\n",
    "    def process(self, element):\n",
    "      return element.split()\n",
    "\n",
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "         | \"Create\" >> beam.Create([\"Hello Beam\", \"This is cool\"])\n",
    "         | 'Print0' >> beam.Map(lambda x: passthrough('Sentence:', x))\n",
    "         | \"Split\" >> beam.ParDo(BreakIntoWordsDoFn()) # Apply a DoFn with a process method\n",
    "         | 'Print1' >> beam.Map(lambda x: passthrough('Word:', x)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "99n6mPVF1C1_"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "         | \"Create\" >> beam.Create([\"Hello Beam\", \"This is cool\"])\n",
    "         | 'Print0' >> beam.Map(lambda x: passthrough('Sentence:', x))\n",
    "         | \"Split\" >> beam.FlatMap(lambda sentence: sentence.split()) # Compare to a FlatMap\n",
    "         | 'Print1' >> beam.Map(lambda x: passthrough('Word:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "kD0CwIu5220t"
   },
   "source": [
    "Again, notice that the results are the same. There is often more than one way to generate the same result!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Gbthg5uU1X4a"
   },
   "source": [
    "## [GroupByKey](https://beam.apache.org/documentation/transforms/python/aggregation/groupbykey/)\n",
    "`GroupByKey` takes a keyed collection of elements and produces a collection where each element consists of a key and all values associated with that key.\n",
    "\n",
    "`GroupByKey` is a transform for processing collections of key/value pairs. It’s a parallel reduction operation, analogous to the Shuffle phase of a Map/Shuffle/Reduce-style algorithm. The input to `GroupByKey` is a collection of key/value pairs that represents a multimap, where the collection contains multiple pairs that have the same key, but different values. Given such a collection, you use `GroupByKey` to collect all of the values associated with each unique key.\n",
    "\n",
    "`GroupByKey` is a good way to aggregate data that has something in common. For example, if you have a collection that stores records of customer orders, you might want to group together all the orders from the same postal code (wherein the “key” of the key/value pair is the postal code field, and the “value” is the remainder of the record)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "PTX7JoS11LQd"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(['apple', 'ball', 'car', 'bear', 'cheetah', 'ant'])\n",
    "           | 'Words' >> beam.Map(lambda x: passthrough('Word:', x))\n",
    "           | beam.Map(lambda word: (word[0], word))\n",
    "           | 'Keyed' >> beam.Map(lambda x: passthrough('Keyed Tuple:', x))\n",
    "           | beam.GroupByKey()\n",
    "           | 'Groups' >> beam.Map(lambda x: passthrough('Grouped:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "DKNvm2H7hrsw"
   },
   "source": [
    "### Now You Try\n",
    "\n",
    "This is a good point to stop reading and try writing code yourself, as a way to help you learn.\n",
    "\n",
    "**Exercise 2 — Group Items by Key**\n",
    "\n",
    "1. Build a Beam pipeline that creates a PCollection containing integers 0 to 10 and prints them.\n",
    "2. Add a step in the pipeline to add a key to each item that will indicate whether it is even or odd.\n",
    "3. Use `GroupByKey` to group even items together and odd items together."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "KwinUX5-hrs0"
   },
   "outputs": [],
   "source": [
    "def passthrough(label, value):\n",
    "    print(label, value)\n",
    "    return value\n",
    "\n",
    "with beam.Pipeline() as pipeline:\n",
    "    lines = (pipeline\n",
    "             | beam.Create(range(11))\n",
    "             | beam.Map(lambda x: passthrough('num', x))\n",
    "             | beam.Map(lambda x: (x%2, x))\n",
    "             | beam.GroupByKey()\n",
    "             | beam.Map(lambda x: passthrough('grouped', x))\n",
    "            )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "9rm6a2W7hrs3"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "3ituEBJzhrs5"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "IqjIVuc1hrs8"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "BFW9cwPMhrtA"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "OkrvfvXohrtB"
   },
   "source": [
    "![Solution](https://urldefense.com/v3/__https://www.tensorflow.org/site-assets/images/marketing/learn/tfx-hero.svg__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxzvhZF7w$ )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "euNLIh_ThrtC"
   },
   "source": [
    "**Solution**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "K3_dXXXIiM1w"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(10 + 1))\n",
    "           | 'Number' >> beam.Map(lambda x: passthrough('Number:', x))\n",
    "           | beam.Map(lambda x: (\"odd\" if x % 2 else \"even\", x))\n",
    "           | 'Even/Odd' >> beam.Map(lambda x: passthrough('Even/Odd:', x))\n",
    "           | beam.GroupByKey()\n",
    "           | 'Groups' >> beam.Map(lambda x: passthrough('Group:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "bsFACrDYgZ9-"
   },
   "source": [
    "`CoGroupByKey` can combine multiple `PCollections`, assuming every element is a tuple whose first item is the key to join on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "G3Mefr2d1fJa"
   },
   "outputs": [],
   "source": [
    "pipeline = beam.Pipeline()\n",
    "\n",
    "# Create two PCollections\n",
    "fruits = pipeline | 'Fruits' >> beam.Create(['apple',\n",
    "                                             'banana',\n",
    "                                             'cherry'])\n",
    "print('fruits is a {}'.format(type(fruits)))\n",
    "\n",
    "countries = pipeline | 'Countries' >> beam.Create(['australia',\n",
    "                                                   'brazil',\n",
    "                                                   'belgium',\n",
    "                                                   'canada'])\n",
    "print('countries is a {}'.format(type(countries)))\n",
    "\n",
    "def add_key(word):\n",
    "    return (word[0], word)\n",
    "\n",
    "# Create PCollections of keyed tuples\n",
    "fruits_with_keys = (fruits | \"fruits_with_keys\" >> beam.Map(add_key)\n",
    "                          | 'Fruit' >> beam.Map(lambda x: passthrough('Fruit:', x)))\n",
    "countries_with_keys = (countries | \"countries_with_keys\" >> beam.Map(add_key)\n",
    "                          | 'Country' >> beam.Map(lambda x: passthrough('Country:', x)))\n",
    "fruitcountries = {\"fruits\": fruits_with_keys, \"countries\": countries_with_keys}\n",
    "\n",
    "# Print the PCollections\n",
    "print('fruits_with_keys is a {}'.format(type(fruits_with_keys)))\n",
    "_ = (fruits_with_keys | 'Print0' >> beam.combiners.ToList()\n",
    "  | beam.Map(lambda x: print('fruits_with_keys:', x)))\n",
    "\n",
    "print('countries_with_keys is a {}'.format(type(countries_with_keys)))\n",
    "_ = (countries_with_keys | 'Print1' >> beam.combiners.ToList()\n",
    "  | beam.Map(lambda x: print('countries_with_keys:', x)))\n",
    "\n",
    "print('fruitcountries: {}'.format(fruitcountries))\n",
    "\n",
    "(fruitcountries | beam.CoGroupByKey() | beam.Map(lambda x: print('CoGrouped:', x)))\n",
    "\n",
    "print('\\nRun the pipeline')\n",
    "pipeline.run().state # Try commenting out this line"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "pLfHkghk3JGd"
   },
   "source": [
    "Try commenting out the last line, and notice that while a `PCollection` is created for the `CoGroupByKey` operation, the pipeline is not actually executed.\n",
    "\n",
    "Also try running the cell above a few times, and note how the order of the print statements changes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "LPi2bXuh5a9l"
   },
   "source": [
    "## [Combine](https://beam.apache.org/documentation/programming-guide/#combine)\n",
    "`Combine` is a transform for combining collections of elements or values. `Combine` has variants that work on entire `PCollections`, and some that combine the values for each key in `PCollections` of key/value pairs.\n",
    "\n",
    "To apply a `Combine` transform, you must provide the function that contains the logic for combining the elements or values. The combining function should be **commutative and associative**, as the function is not necessarily invoked exactly once on all values with a given key. Because the input data (including the value collection) **may be distributed across multiple workers**, the combining function might be called multiple times to perform partial combining on subsets of the value collection. The Beam SDK also provides some pre-built combine functions for common numeric combination operations such as `sum`, `min`, and `max`.\n",
    "\n",
    "Simple combine operations, such as sums, can usually be implemented as a simple function. More complex combination operations might require you to create a subclass of `CombineFn` that has an accumulation type distinct from the input/output type.\n",
    "\n",
    "Try it first with a sum."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "mWC8ywHC5KQ9"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create([1, 2, 3, 4, 5])\n",
    "           | 'Numbers' >> beam.Map(lambda x: passthrough('Number:', x))\n",
    "           | beam.CombineGlobally(sum)\n",
    "           | 'Sums' >> beam.Map(lambda x: passthrough('Sum:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "B9Jsahga-YSE"
   },
   "source": [
    "Now try calculating the mean using the pre-built combine function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "DS58V5cd5pQH"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create([1, 2, 3, 4, 5])\n",
    "           | 'Numbers' >> beam.Map(lambda x: passthrough('Number:', x))\n",
    "           | beam.combiners.Mean.Globally()\n",
    "           | 'Means' >> beam.Map(lambda x: passthrough('Mean:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "m0yQwqUB-r8N"
   },
   "source": [
    "Now try creating a custom `CombineFn` to calculate the mean. Note the required concrete implementations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "1R9zt-xX5h-Y"
   },
   "outputs": [],
   "source": [
    "class AverageFn(beam.CombineFn):\n",
    "  def create_accumulator(self):\n",
    "    return (0.0, 0)\n",
    "  \n",
    "  def add_input(self, accumulator, input_):\n",
    "    total, count = accumulator\n",
    "    total += input_\n",
    "    count += 1\n",
    "    return (total, count)\n",
    "\n",
    "  def merge_accumulators(self, accumulators):\n",
    "    totals, counts = zip(*accumulators)\n",
    "    return sum(totals), sum(counts)\n",
    "\n",
    "  def extract_output(self, accumulator):\n",
    "    total, count = accumulator\n",
    "    return total / count if count else float(\"NaN\")\n",
    "\n",
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create([1, 2, 3, 4, 5])\n",
    "           | 'Numbers' >> beam.Map(lambda x: passthrough('Number:', x))\n",
    "           | beam.CombineGlobally(AverageFn())\n",
    "           | 'Means' >> beam.Map(lambda x: passthrough('Mean:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "a4VBhAcc_pt_"
   },
   "source": [
    "Now try a simple `Count` using the pre-built combine function. This one does a `PerElement` count."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "9N0So44g6_8M"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(['bob', 'alice', 'alice', 'bob', 'charlie', 'alice'])\n",
    "           | 'Persons' >> beam.Map(lambda x: passthrough('Person:', x))\n",
    "           | beam.combiners.Count.PerElement()\n",
    "           | 'Counts' >> beam.Map(lambda x: passthrough('Count:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "LI3BEiBIANHl"
   },
   "source": [
    "We can also create a count by using `CombinePerKey` with the Python `sum` function, using the people as keys."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Vafplz7d56CF"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(['bob', 'alice', 'alice', 'bob', 'charlie', 'alice'])\n",
    "           | 'Persons' >> beam.Map(lambda x: passthrough('Person:', x))\n",
    "           | beam.Map(lambda word: (word, 1))\n",
    "           | 'Keyed' >> beam.Map(lambda x: passthrough('Key:', x))\n",
    "           | beam.CombinePerKey(sum)\n",
    "           | 'Counts' >> beam.Map(lambda x: passthrough('Count:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "B6KyLL8wA4EO"
   },
   "source": [
    "Now try a simple `Count` using the pre-built combine function. This one does a count `Globally`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "lswZZCAe6Th5"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(['bob', 'alice', 'alice', 'bob', 'charlie', 'alice'])\n",
    "           | 'Persons' >> beam.Map(lambda x: passthrough('Person:', x))\n",
    "           | beam.combiners.Count.Globally()\n",
    "           | 'Counts' >> beam.Map(lambda x: passthrough('Count:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "tgoLTv8gi0WW"
   },
   "source": [
    "### Now You Try\n",
    "\n",
    "This is a good point to stop reading and try writing code yourself, as a way to help you learn.\n",
    "\n",
    "**Exercise 3 — Combine Items**\n",
    "\n",
    "1. Start with Beam pipeline you built in the previous exercise: it creates a `PCollection` containing integers 0 to 10, and labels them as odd or even.\n",
    "2. Add another step to make the pipeline compute the square of each number.\n",
    "3. Add a step that groups all of the odd or even numbers and computes the mean of each group (i.e., the mean of all odd numbers between 0 and 10, and the mean of all even numbers between 0 and 10). You can use the `AverageFn` that we created above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Cjw3Sqjci0Wd"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "    result = (\n",
    "        pipeline\n",
    "        | 'Create Array' >> beam.Create(range(11))\n",
    "        | 'Square Element' >> beam.Map(lambda x: x**2)\n",
    "        | 'Generate Key' >> beam.Map(lambda x: (x%2, x))\n",
    "        | 'Average by Key' >> beam.CombinePerKey(AverageFn())\n",
    "        | 'Print' >> beam.Map(lambda x: passthrough('Avg', x))\n",
    "    )\n",
    "# 0 2 4 6 8 10\n",
    "# 1 3 5 7 9\n",
    "display_pipeline(pipeline)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "AalDelXSi0Wh"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "EPb9-f9Ri0Wj"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "LQ3EL2QFi0Wl"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "pgbf6O42i0Wo"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "6SzSYQ2ei0Wp"
   },
   "source": [
    "![Solution](https://urldefense.com/v3/__https://www.tensorflow.org/site-assets/images/marketing/learn/tfx-hero.svg__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxzvhZF7w$ )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "qMYzcG46i0Wq"
   },
   "source": [
    "**Solution**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "KSOWJXA_jt1r"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(10 + 1))\n",
    "           | 'Numbers' >> beam.Map(lambda x: passthrough('Number:', x))\n",
    "           | 'Label OddEven' >> beam.Map(lambda x: (\"odd\" if x % 2 else \"even\", x))\n",
    "           | 'Print OddEven' >> beam.Map(lambda x: passthrough('OddEven:', x))\n",
    "           | 'Square' >> beam.Map(lambda x: (x[0], x[1] ** 2))\n",
    "           | 'Print Squares' >> beam.Map(lambda x: passthrough('Squared:', x))\n",
    "           | 'Calculate Means' >> beam.CombinePerKey(AverageFn())\n",
    "           | 'Print Means' >> beam.Map(lambda x: passthrough('Mean:', x)))\n",
    "\n",
    "display_pipeline(pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "n-SEoKnd8cea"
   },
   "source": [
    "## [Flatten](https://beam.apache.org/documentation/programming-guide/#flatten)\n",
    "`Flatten` is a transform for `PCollection` objects that store **the same data type**. `Flatten` merges multiple `PCollection` objects into a single logical `PCollection`.\n",
    "\n",
    "#### Data encoding in merged collections\n",
    "By default, the coder for the output `PCollection` is the same as the coder for the first `PCollection` in the input `PCollectionList`. However, the input `PCollection` objects can each use different coders, **as long as they all contain the same data type** in your chosen language.\n",
    "\n",
    "#### Merging windowed collections\n",
    "When using `Flatten` to merge `PCollection` objects that have a windowing strategy applied, all of the `PCollection` objects you want to merge must use a compatible windowing strategy and window sizing. For example, all the collections you're merging must all use (hypothetically) identical 5-minute fixed windows or 4-minute sliding windows starting every 30 seconds.\n",
    "\n",
    "If your pipeline attempts to use `Flatten` to merge `PCollection` objects with incompatible windows, Beam generates an `IllegalStateException` error when your pipeline is constructed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "0BVhknBM7Frz"
   },
   "outputs": [],
   "source": [
    "pipeline = beam.Pipeline()\n",
    "\n",
    "wordsStartingWithA = pipeline | 'Words starting with A' >> beam.Create(['apple', 'ant', 'arrow'])\n",
    "\n",
    "# Note the required parens for multi-line\n",
    "wordsStartingWithB = (pipeline\n",
    "  | 'Words starting with B' >> beam.Create(['ball', 'book', 'bow']))\n",
    "\n",
    "result = (wordsStartingWithA, wordsStartingWithB) | beam.Flatten()\n",
    "\n",
    "# Print the PCollections\n",
    "_ = (wordsStartingWithA | 'PrintListA' >> beam.combiners.ToList()\n",
    "  | 'Print A' >> beam.Map(lambda x: passthrough('wordsStartingWithA:', x)))\n",
    "_ = (wordsStartingWithB | 'PrintListB' >> beam.combiners.ToList()\n",
    "  | 'Print B' >> beam.Map(lambda x: passthrough('wordsStartingWithB:', x)))\n",
    "_ = (result | 'Result' >> beam.combiners.ToList()\n",
    "  | 'Print Result' >> beam.Map(lambda x: passthrough('result:', x)))\n",
    "\n",
    "pipeline.run().state\n",
    "display_pipeline(pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "etCS1bQlQNJx"
   },
   "source": [
    "Try running the cell above a few times, and note how the order of the print statements changes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "rvaC0Zvj9Cse"
   },
   "source": [
    "## [Partition](https://beam.apache.org/documentation/programming-guide/#partition)\n",
    "`Partition` is a transform for `PCollection` objects that store **the same data type**. `Partition` splits a single `PCollection` into a fixed number of smaller collections.\n",
    "\n",
    "`Partition` divides the elements of a `PCollection` according to a **partitioning function** that you provide. The partitioning function contains the logic that determines how to split up the elements of the input `PCollection` into each resulting partition `PCollection`. The number of partitions must be determined at graph construction time. You can, for example, pass the number of partitions as a command-line option at runtime (which will then be used to build your pipeline graph), but you cannot determine the number of partitions in mid-pipeline (based on data calculated after your pipeline graph is constructed, for instance).\n",
    "\n",
    "Let's partition a group of numbers in ranges of 100."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "P6wjB7Y48olW"
   },
   "outputs": [],
   "source": [
    "def partition_fn(number, num_partitions):\n",
    "    partition = number // 100\n",
    "    return min(partition, num_partitions - 1)\n",
    "\n",
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create([1, 110, 2, 350, 4, 5, 100, 150, 3])\n",
    "           | beam.Partition(partition_fn, 3))\n",
    "  \n",
    "  # Print the PCollections\n",
    "  _ = (lines[0] | 'PrintSmallList' >> beam.combiners.ToList()\n",
    "    | 'Print Small' >> beam.Map(lambda x: passthrough('Small:', x)))\n",
    "  _ = (lines[1] | 'PrintMediumList' >> beam.combiners.ToList()\n",
    "    | 'Print Medium' >> beam.Map(lambda x: passthrough('Medium:', x)))\n",
    "  _ = (lines[2] | 'PrintLargeList' >> beam.combiners.ToList()\n",
    "    | 'Print Large' >> beam.Map(lambda x: passthrough('Large:', x)))\n",
    "\n",
    "display_pipeline(pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "kBiNHmlcFBb1"
   },
   "source": [
    "## [Side Inputs](https://beam.apache.org/documentation/programming-guide/#side-inputs)\n",
    "In addition to the main input `PCollection`, you can provide additional inputs to a `ParDo` transform in the form of side inputs. A side input is an additional input that your `DoFn` can access each time it processes an element in the input `PCollection`. When you specify a side input, you create a view of some other data that can be read from within the `ParDo` transform’s `DoFn` while processing each element.\n",
    "\n",
    "Side inputs are useful if your `ParDo` needs to inject additional data when processing each element in the input PCollection, but the additional data needs to be determined at runtime (and not hard-coded). Such values might be determined by the input data, or depend on a different branch of your pipeline.\n",
    "\n",
    "Let's use a side input to increment numbers by a configurable amount. First, we'll use the default for our `ParDo` transform."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "YTZWcmdB9eB9"
   },
   "outputs": [],
   "source": [
    "def increment(number, inc=1):\n",
    "    return number + inc\n",
    "\n",
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | \"Create\" >> beam.Create([1, 2, 3, 4, 5])\n",
    "           | \"Increment\" >> beam.Map(increment)\n",
    "           | \"Print\" >> beam.ParDo(print))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "MKg0SV4ja4A1"
   },
   "source": [
    "Next, we'll pass a number to increment by."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Men6SXjGFkKH"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | \"Create\" >> beam.Create([1, 2, 3, 4, 5])\n",
    "           | \"Increment\" >> beam.Map(increment, 10) # Pass a side input of 10\n",
    "           | \"Print\" >> beam.ParDo(print))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "j2LAoh1fupfp"
   },
   "source": [
    "## [Additional Outputs](https://beam.apache.org/documentation/programming-guide/#additional-outputs)\n",
    "While `ParDo` always produces a main output `PCollection` (as the return value from `apply`), you can also have your `ParDo` produce any number of additional output `PCollections`. If you choose to have multiple outputs, your `ParDo` returns all of the output `PCollections` (including the main output) bundled together.\n",
    "\n",
    "To emit elements to multiple output `PCollections`, invoke `with_outputs()` on the `ParDo`, and specify the expected tags for the outputs. `with_outputs()` returns a `DoOutputsTuple` object. Tags specified in `with_outputs` are attributes on the returned `DoOutputsTuple` object. The tags give access to the corresponding output `PCollections`.\n",
    "\n",
    "Let's try creating two `PCollections`, one each for odd and even numbers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "hlv1qnPeFt54"
   },
   "outputs": [],
   "source": [
    "def compute(number):\n",
    "  if number % 2 == 0:\n",
    "      yield number\n",
    "  else:\n",
    "      yield pvalue.TaggedOutput(\"odd\", number + 10)\n",
    "\n",
    "with beam.Pipeline() as pipeline:\n",
    "  even, odd = (pipeline\n",
    "           | \"Create\" >> beam.Create([1, 2, 3, 4, 5, 6, 7])\n",
    "           | \"Increment\" >> beam.ParDo(compute).with_outputs(\"odd\", main=\"even\"))\n",
    "  \n",
    "  # Print the PCollections\n",
    "  _ = (even | 'PrintEvenList' >> beam.combiners.ToList()\n",
    "    | 'Print Even' >> beam.Map(lambda x: passthrough('Evens:', x)))\n",
    "  _ = (odd | 'PrintOddList' >> beam.combiners.ToList()\n",
    "    | 'Print Odd' >> beam.Map(lambda x: passthrough('Odds:', x)))\n",
    "\n",
    "display_pipeline(pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "jTubowaGcu4J"
   },
   "source": [
    "Note how the `Increment` stage returns two different `PCollections` in the graph above. **Untagged elements** will always be placed in the \"main\" `PCollection`.\n",
    "\n",
    "Try tagging even elements and see what happens!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ZYFrovl8IGsB"
   },
   "source": [
    "## [Branching](https://beam.apache.org/documentation/programming-guide/#applying-transforms)\n",
    "A transform does not consume or otherwise alter the input collection – remember that a `PCollection` is immutable by definition. This means that you can apply multiple transforms to the same input `PCollection` to create a branching pipeline.\n",
    "\n",
    "Let's create a pipeline with 3 branches, which produces 3 different `PCollections`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "cc9hrASXGmug"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as branching_pipeline:\n",
    "  numbers = (branching_pipeline | beam.Create([1, 2, 3, 4, 5]))\n",
    "  mult5_results = numbers | 'Multiply by 5' >> beam.Map(lambda num: num * 5)\n",
    "  mult10_results = numbers | 'Multiply by 10' >> beam.Map(lambda num: num * 10)\n",
    "\n",
    "  # Print the PCollections\n",
    "  _ = (numbers | 'PrintNumbersList' >> beam.combiners.ToList()\n",
    "    | 'Print Numbers' >> beam.Map(lambda x: passthrough('Numbers:', x)))\n",
    "  _ = (mult5_results | 'Print5List' >> beam.combiners.ToList()\n",
    "    | 'Print Mult 5' >> beam.Map(lambda x: passthrough('Mult 5:', x)))\n",
    "  _ = (mult10_results | 'Print10List' >> beam.combiners.ToList()\n",
    "    | 'Print Mult 10' >> beam.Map(lambda x: passthrough('Mult 10:', x)))\n",
    "\n",
    "display_pipeline(branching_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1-_FQ3S9IotY"
   },
   "source": [
    "## [Composite Transforms](https://beam.apache.org/documentation/programming-guide/#composite-transforms)\n",
    "Transforms can have a nested structure, where a complex transform performs multiple simpler transforms (such as more than one `ParDo`, `Combine`, `GroupByKey`, or even other composite transforms). These transforms are called ***composite transforms***. Nesting multiple transforms inside a single composite transform can make your code more modular and easier to understand.\n",
    "\n",
    "Your composite transform's parameters and return value must match the initial input type and final return type for the entire transform, even if the transform's intermediate data changes type multiple times.\n",
    "\n",
    "To create a composite transform, create a subclass of the `PTransform` class and override the `expand` method to specify the actual processing logic. Then use this transform just as you would a built-in transform."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "I8GplhSxIY8I"
   },
   "outputs": [],
   "source": [
    "# Subclass beam.PTransform to create a composite transform\n",
    "class ExtractAndMultiplyNumbers(beam.PTransform):\n",
    "    def expand(self, pcollection):\n",
    "        return (pcollection\n",
    "            | beam.FlatMap(lambda line: line.split(\",\"))\n",
    "            | beam.Map(lambda num: int(num) * 10))\n",
    "\n",
    "with beam.Pipeline() as composite_pipeline:\n",
    "  number_PCollection = (composite_pipeline\n",
    "           | beam.Create(['1,2,3,4,5', '6,7,8,9,10'])\n",
    "           | ExtractAndMultiplyNumbers())\n",
    "  \n",
    "  # Print the PCollection\n",
    "  _ = (number_PCollection | 'PrintNumbersList' >> beam.combiners.ToList()\n",
    "    | 'Print Numbers' >> beam.Map(lambda x: passthrough('Numbers:', x)))\n",
    "\n",
    "display_pipeline(composite_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "b7NFpWUMJIJy"
   },
   "source": [
    "## [Filter](https://beam.apache.org/documentation/transforms/python/elementwise/filter/)\n",
    "`Filter`, given a predicate, filters out all elements that don't satisfy that predicate. `Filter` may also be used to filter based on an inequality with a given value based on the comparison ordering of the element.  You can pass functions with multiple arguments to `Filter`. They are passed as additional positional arguments or keyword arguments to the function.  If the `PCollection` has a single value, such as the average from another computation, passing the `PCollection` as a *singleton* accesses that value.  If the `PCollection` has multiple values, pass the `PCollection` as an *iterator*. This accesses elements lazily as they are needed, so it is possible to iterate over large `PCollections` that won't fit into memory.\n",
    "\n",
    "> Note: You can pass the `PCollection` as a list with `beam.pvalue.AsList(PCollection)`, but this requires that all the elements fit into memory.\n",
    "\n",
    "> Note: You can pass the `PCollection` as a dictionary with `beam.pvalue.AsDict(PCollection)`. Each element must be a (key, value) pair. Note that all the elements of the `PCollection` must fit into memory.\n",
    "\n",
    "If the `PCollection` won't fit into memory, use `beam.pvalue.AsIter(PCollection)` instead.\n",
    "\n",
    "First, let's try filtering out even numbers using a `ParDo` transform."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "BuFN36jVI4gH"
   },
   "outputs": [],
   "source": [
    "class FilterOddNumbers(beam.DoFn):\n",
    "    def process(self, element, *args, **kwargs):\n",
    "        if element % 2 == 1:\n",
    "            yield element\n",
    "\n",
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(1, 11))\n",
    "           | beam.ParDo(FilterOddNumbers())\n",
    "           | beam.Map(lambda x: passthrough('Odd number:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "hqlPQSCwTsFS"
   },
   "source": [
    "Next, let's use `Filter` to do the same thing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "8zQlSBJWJYrY"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(1, 11))\n",
    "           | beam.Filter(lambda num: num % 2 == 1)\n",
    "           | beam.Map(lambda x: passthrough('Odd number:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Ic7kHP5sJphJ"
   },
   "source": [
    "## [Aggregation](https://beam.apache.org/documentation/programming-guide/)\n",
    "Beam uses [windowing](https://urldefense.com/v3/__https://beam.apache.org/documentation/programming-guide/*windowing__;Iw!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxDyJyXgQ$ ) to divide a continuously updating unbounded `PCollection` into logical windows of finite size. These logical windows are determined by some characteristic associated with a data element, such as a timestamp. [Aggregation transforms](https://urldefense.com/v3/__https://beam.apache.org/documentation/transforms/python/overview/*aggregation__;Iw!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxGg6bt2E$ ) (such as GroupByKey and Combine) work on a per-window basis — as the data set is generated, they process each `PCollection` as a succession of these finite windows.\n",
    "\n",
    "A related concept, called [triggers](https://urldefense.com/v3/__https://beam.apache.org/documentation/programming-guide/*triggers__;Iw!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxSPrnvUA$ ), determines when to emit the results of aggregation as unbounded data arrives. You can use triggers to refine the windowing strategy for your `PCollection`. Triggers allow you to deal with late-arriving data, or to provide early results.\n",
    "\n",
    "First, let's count the elements in a `PCollection`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "O3W-h0AzJg3f"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(1, 11))\n",
    "           | beam.combiners.Count.Globally() # Count\n",
    "           | beam.Map(lambda x: passthrough('Count =', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "TMHW9Je3VCUP"
   },
   "source": [
    "Next, let's sum the elements with a `CombineGlobally` aggregator and Python's sum."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "GZVWg2AvKxUz"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(1, 11))\n",
    "           | beam.CombineGlobally(sum) # CombineGlobally sum\n",
    "           | beam.Map(lambda x: passthrough('Sum =', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "tRXBNpUTVkRw"
   },
   "source": [
    "Next, let's calculate the mean."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "PpkdeF6YK2gc"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(1, 11))\n",
    "           | beam.combiners.Mean.Globally() # Mean\n",
    "           | beam.Map(lambda x: passthrough('Mean =', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Zkw5rjkEVv7M"
   },
   "source": [
    "Next, let's return the smallest element in the `PCollection`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "EYJCGAgdK227"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(1, 11))\n",
    "           | beam.combiners.Top.Smallest(1) # Top Smallest\n",
    "           | beam.Map(lambda x: passthrough('Smallest =', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "XnQpGZTYV5t-"
   },
   "source": [
    "Next, let's return the largest element in the `PCollection`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "aXySuL7QK4sb"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.Create(range(1, 11))\n",
    "           | beam.combiners.Top.Largest(1) # Top Largest\n",
    "           | beam.Map(lambda x: passthrough('Largest =', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "GMKY3LCWLGXu"
   },
   "source": [
    "# [Pipeline I/O](https://beam.apache.org/documentation/programming-guide/#pipeline-io)\n",
    "When you create a pipeline, you often need to read data from some external source, such as a file or a database. Likewise, you may want your pipeline to output its result data to an external storage system. Beam provides read and write transforms for a [number of common data storage types](https://beam.apache.org/documentation/io/built-in/). If you want your pipeline to read from or write to a data storage format that isn’t supported by the built-in transforms, you can [implement your own read and write transforms](https://beam.apache.org/documentation/io/developing-io-overview/) \n",
    "### Download example data\n",
    "Download the sample dataset for use with the cells below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "BywX6OUEhAqn"
   },
   "outputs": [],
   "source": [
    "import os, tempfile, urllib\n",
    "DATA_PATH = 'https://raw.githubusercontent.com/tensorflow/tfx/master/tfx/examples/chicago_taxi_pipeline/data/simple/data.csv'\n",
    "_data_root = tempfile.mkdtemp(prefix='tfx-data')\n",
    "_data_filepath = os.path.join(_data_root, \"data.csv\")\n",
    "urllib.request.urlretrieve(DATA_PATH, _data_filepath)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4pwOXlPQYiJn"
   },
   "source": [
    "Let's take a quick look at the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Hqn4wST2Bex5"
   },
   "outputs": [],
   "source": [
    "!head {_data_filepath}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "R9BQpD-JYlte"
   },
   "source": [
    "Let's filter out only the lines that we're interested in."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "hz4oEI90mB-m"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.io.ReadFromText(_data_filepath)\n",
    "           | beam.Filter(lambda line: line.startswith(\"13,16.45,11,12,3\"))\n",
    "           | beam.Map(lambda x: passthrough('Match:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "8siWql-89oq1"
   },
   "source": [
    "### Putting Everything Together\n",
    "\n",
    "Use several of the concepts, classes, and methods discussed above in a concrete example.\n",
    "\n",
    "**Exercise 4 — Reading, Filtering, Parsing, Grouping and Averaging**\n",
    "\n",
    "Write a Beam pipeline that reads the dataset, computes the mean fare for each company, and takes the top 5 companies with the largest mean fares.\n",
    "\n",
    "*Hints*:\n",
    "* Use the code above to read the dataset.\n",
    "* Add a `Map` step to split each row on the commas.\n",
    "* Filter out the header row and any rows without values for company.\n",
    "* Calculate the mean fare for each company\n",
    "* (Optional) Print the mean fares for each company\n",
    "* Take the top 5 companies by mean fare\n",
    "* Print the list of the top 5 companies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "I_jeg125l6sR"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "    average_per_company = (\n",
    "        pipeline\n",
    "        | beam.io.ReadFromText(_data_filepath)\n",
    "        | beam.Map(lambda x: x.split(','))\n",
    "        | beam.Filter(lambda x: x[0] != 'pickup_community_area' and x[-4])\n",
    "        | beam.Map(lambda x: (x[-4], float(x[1])))\n",
    "        | beam.CombinePerKey(AverageFn())\n",
    "    )\n",
    "    \n",
    "    print_average_fare = (\n",
    "        average_per_company\n",
    "        | beam.Map(lambda x: passthrough('avg({}) ='.format(x[0]), x[1]))\n",
    "    )\n",
    "    \n",
    "    print_average_top5 = (\n",
    "        average_per_company\n",
    "        | beam.Map(lambda x: (x[1], x[0]))\n",
    "        | beam.combiners.Top.Largest(5)\n",
    "        | beam.Map(lambda x: passthrough('Top 5 Companies =', x))\n",
    "    )\n",
    "    \n",
    "display_pipeline(pipeline)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "3q6OTO5Xl6sU"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "U5sVJaojl6sW"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Y7LfH_rHl6sZ"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "lNSVxnmnl6sa"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "u8pjA5uFl6sc"
   },
   "source": [
    "![Solution](https://urldefense.com/v3/__https://www.tensorflow.org/site-assets/images/marketing/learn/tfx-hero.svg__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxzvhZF7w$ )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "I2mfIjMbl6sc"
   },
   "source": [
    "**Solution**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "QGLY9QRJLB8y"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as pipeline:\n",
    "  lines = (pipeline\n",
    "           | beam.io.ReadFromText(_data_filepath)\n",
    "           | beam.Map(lambda line: line.split(\",\")) # CSV\n",
    "           | beam.Filter(lambda cols: len(cols[-4]) > 0 and cols[0] != 'pickup_community_area')\n",
    "           | beam.Map(lambda cols: (cols[-4], float(cols[1]))) # (Company, Mean Fare) for PerKey\n",
    "           | beam.combiners.Mean.PerKey() # Mean fare by company\n",
    "           | beam.Map(lambda cols: (cols[1], cols[0])) # Switch to (Mean Fare, Company) for Top\n",
    "           | beam.Map(lambda x: passthrough('(Mean Fare, Company):', x))\n",
    "           | beam.combiners.Top.Largest(5) # Top 5 mean fares\n",
    "           | beam.Map(lambda x: passthrough('Top 5:', x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "SZlIAvT4o0tW"
   },
   "source": [
    "## Beam I/O and TFX\n",
    "\n",
    "Since TensorFlow and TFX often use the [TFRecord](https://urldefense.com/v3/__https://www.tensorflow.org/tutorials/load_data/tfrecord__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxBE9fjf0$ ) format, let's use Beam I/O with TFRecord files.  \n",
    "\n",
    "First, let's create a test file containing 10 records of \"Record X\" strings in UTF-8."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T14:59:53.133720Z",
     "start_time": "2021-04-13T14:59:49.687013Z"
    },
    "id": "IGEzjUGaRHCW"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Record 0\n",
      "Record 1\n",
      "Record 2\n",
      "Record 3\n",
      "Record 4\n",
      "Record 5\n",
      "Record 6\n",
      "Record 7\n",
      "Record 8\n",
      "Record 9\n"
     ]
    }
   ],
   "source": [
    "#!pip install -q -U \"tensorflow>=2.1,<=2.2\"\n",
    "\n",
    "import tensorflow as tf\n",
    "with tf.io.TFRecordWriter(\"test.tfrecord\") as tfrecord_file:\n",
    "  for index in range(10):\n",
    "    tfrecord_file.write(\"Record {}\".format(index).encode(\"utf-8\"))\n",
    "\n",
    "dataset = tf.data.TFRecordDataset('test.tfrecord')\n",
    "for record in dataset:\n",
    "  print(record.numpy().decode('utf-8'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "FzAt26WBqgKE"
   },
   "source": [
    "Now let's read the test file and process each of the records.  We'll write the results to a second `TFRecord` file, one record for each record processed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-04-13T15:00:07.402486Z",
     "start_time": "2021-04-13T15:00:07.131336Z"
    },
    "id": "hgRAt4nRR-qf"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "test_processed.tfrecord-00000-of-00001\n"
     ]
    }
   ],
   "source": [
    "with beam.Pipeline() as rw_pipeline:\n",
    "  lines = (rw_pipeline\n",
    "           | 'Read test file' >> beam.io.ReadFromTFRecord(\"test.tfrecord\")\n",
    "           | 'Process each record' >> beam.Map(lambda line: line + b' processed')\n",
    "           | 'Write the results to a new file' >> beam.io.WriteToTFRecord(\"test_processed.tfrecord\")\n",
    "           | 'Print the filename' >> beam.ParDo(print))\n",
    "\n",
    "#display_pipeline(rw_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "X_j06O0ktIdy"
   },
   "source": [
    "Now let's read the results file and print each record."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "sT_bo1VxTbTb"
   },
   "outputs": [],
   "source": [
    "with beam.Pipeline() as utf_pipeline:\n",
    "  lines = (utf_pipeline\n",
    "           | \"Read the results file\" >> beam.io.ReadFromTFRecord(\"test_processed.tfrecord-00000-of-00001\")\n",
    "           | \"Decode UTF-8\" >> beam.Map(lambda line: line.decode('utf-8'))\n",
    "           | \"Print each record\" >> beam.ParDo(print))\n",
    "\n",
    "display_pipeline(utf_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "KCAysNHLQK80"
   },
   "source": [
    "Note that there are many [other built-in I/O transforms](https://urldefense.com/v3/__https://beam.apache.org/documentation/io/built-in/__;!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxE3x8M-c$ )."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "KZDYehqpPpXY"
   },
   "source": [
    "# [Windowing](https://beam.apache.org/documentation/programming-guide/#windowing)\n",
    "As discussed above, windowing subdivides a `PCollection` according to the timestamps, or some other ordering or grouping, of its individual elements.\n",
    "\n",
    "Some Beam transforms, such as `GroupByKey` and `Combine`, group multiple elements by a common key. Ordinarily, that grouping operation groups all of the elements that have the same key within the entire data set. With an unbounded data set, it is impossible to collect all of the elements, since new elements are constantly being added and may be infinitely many (e.g. streaming data). If you are working with unbounded `PCollections`, windowing is especially useful.\n",
    "\n",
    "In the Beam model, any `PCollection` (including unbounded `PCollections`) can be subdivided into logical windows. Each element in a `PCollection` is assigned to one or more windows according to the `PCollection`'s windowing function, and each individual window contains a finite number of elements. Grouping transforms then consider each `PCollection`'s elements on a per-window basis. `GroupByKey`, for example, implicitly groups the elements of a `PCollection` by key and window.\n",
    "\n",
    "Additional information on Beam Windowing is available in the [Beam Programming Guide](https://urldefense.com/v3/__https://beam.apache.org/documentation/programming-guide/*windowing__;Iw!!PIZeeW5wscynRQ!55X7LkP-oqeRJ6UUmtPVv4vn3HLBOqlFNpBPTKq9j6A_FsUVIeCO-4RxDyJyXgQ$ )."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "tLMh3chYC1sE"
   },
   "source": [
    "Let's begin by creating a pipeline to:\n",
    "\n",
    "* Read our CSV file above\n",
    "* Create fixed windows for each week\n",
    "* Calculate the mean fare by company for each week\n",
    "* Group the companies by week\n",
    "* Print out 2016 only"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Qd3a4hwlVqfF"
   },
   "outputs": [],
   "source": [
    "import datetime\n",
    "\n",
    "DAYS = 24 * 60 * 60\n",
    "\n",
    "class BuildRecordFn(beam.DoFn):\n",
    "  def process(self, element,  window=beam.DoFn.WindowParam):\n",
    "      window_end = str(window.end.to_utc_datetime())\n",
    "      return [(window_end, element)]\n",
    "\n",
    "class AssignTimestamps(beam.DoFn):\n",
    "  def process(self, element):\n",
    "    ts = int(element[5])\n",
    "    yield beam.window.TimestampedValue(element, ts)\n",
    "\n",
    "with beam.Pipeline() as window_pipeline:\n",
    "  lines = (window_pipeline\n",
    "           | 'Read text file' >> beam.io.ReadFromText(_data_filepath)\n",
    "           | 'Parse CSV' >> beam.Map(lambda line: line.split(\",\"))\n",
    "           | 'Filter out' >> beam.Filter(lambda cols: len(cols[-4]) > 0 and cols[0] != 'pickup_community_area')\n",
    "           | 'Assign timestamps' >> beam.ParDo(AssignTimestamps())\n",
    "           | 'Create week windows' >> beam.WindowInto(beam.window.FixedWindows(7*DAYS))\n",
    "           | '(Company, Mean Fare) for PerKey' >> beam.Map(lambda cols: (cols[-4], float(cols[1])))\n",
    "           | 'Mean fare by company' >> beam.combiners.Mean.PerKey()\n",
    "           | 'Add window end' >> beam.ParDo(BuildRecordFn())\n",
    "           | 'Group windows' >> beam.GroupByKey()\n",
    "           | 'Only include 2016' >> beam.Filter(lambda win: win[0] > '2016' and win[0] < '2017')\n",
    "#           | 'Sort by Key' >> beam.SortByKey()\n",
    "           | 'Print' >> beam.Map(lambda x: passthrough('(Window end, [Mean fares by company]):', x)))\n",
    "\n",
    "display_pipeline(window_pipeline)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [
    "R_vFbXN2T7oo"
   ],
   "name": "Developing With Apache Beam",
   "private_outputs": true,
   "provenance": [],
   "toc_visible": true
  },
  "environment": {
   "name": "tf2-2-3-gpu.2-3.m55",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/tf2-2-3-gpu.2-3:m55"
  },
  "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"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}