Word_Count_in_Apache_Beam.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Word Count in Apache Beam",
      "provenance": [],
      "collapsed_sections": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python2",
      "display_name": "Python 2"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/thecodemancer/Apache-Beam/blob/main/Word_Count_in_Apache_Beam.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "metadata": {
        "id": "lNKIMlEDZ_Vw"
      },
      "cell_type": "markdown",
      "source": [
        "# Word Count in Apache Beam\n",
        "\n",
        "In this notebook, we set up your development environment and work through a simple example using the [DirectRunner](https://beam.apache.org/documentation/runners/direct/)."
      ]
    },
    {
      "metadata": {
        "id": "Fz6KSQ13_3Rr"
      },
      "cell_type": "markdown",
      "source": [
        "# Setup\n",
        "\n",
        "First, you need to set up your environment, which includes installing `apache-beam` and downloading a text file from Cloud Storage to your local file system. We are using this file to test your pipeline."
      ]
    },
    {
      "metadata": {
        "id": "GOOk81Jj_yUy",
        "outputId": "9d2fae95-2800-45c7-a41c-f38be365c8b4",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "cell_type": "code",
      "source": [
        "# Run and print a shell command.\n",
        "def run(cmd):\n",
        "  print('>> {}'.format(cmd))\n",
        "  !{cmd}\n",
        "  print('')\n",
        "\n",
        "# Install apache-beam.\n",
        "run('pip install --quiet apache-beam')\n",
        "\n",
        "# Copy the input file into the local file system.\n",
        "run('mkdir -p data')\n",
        "run('gsutil cp gs://dataflow-samples/shakespeare/kinglear.txt data/')"
      ],
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            ">> pip install --quiet apache-beam\n",
            "\u001b[K     |████████████████████████████████| 10.9 MB 7.1 MB/s \n",
            "\u001b[K     |████████████████████████████████| 151 kB 68.1 MB/s \n",
            "\u001b[K     |████████████████████████████████| 508 kB 19.0 MB/s \n",
            "\u001b[K     |████████████████████████████████| 2.4 MB 59.4 MB/s \n",
            "\u001b[K     |████████████████████████████████| 62 kB 1.6 MB/s \n",
            "\u001b[K     |████████████████████████████████| 46 kB 2.1 MB/s \n",
            "\u001b[K     |████████████████████████████████| 275 kB 73.3 MB/s \n",
            "\u001b[K     |████████████████████████████████| 1.0 MB 64.7 MB/s \n",
            "\u001b[?25h  Building wheel for dill (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Building wheel for docopt (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "\u001b[31mERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.\n",
            "tensorflow 2.8.2+zzzcolab20220719082949 requires protobuf<3.20,>=3.9.2, but you have protobuf 3.20.1 which is incompatible.\n",
            "gym 0.17.3 requires cloudpickle<1.7.0,>=1.2.0, but you have cloudpickle 2.1.0 which is incompatible.\u001b[0m\n",
            "\n",
            ">> mkdir -p data\n",
            "\n",
            ">> gsutil cp gs://dataflow-samples/shakespeare/kinglear.txt data/\n",
            "Copying gs://dataflow-samples/shakespeare/kinglear.txt...\n",
            "/ [1 files][153.6 KiB/153.6 KiB]                                                \n",
            "Operation completed over 1 objects/153.6 KiB.                                    \n",
            "\n"
          ]
        }
      ]
    },
    {
      "metadata": {
        "id": "cPvvFB19uXNw"
      },
      "cell_type": "markdown",
      "source": [
        "# Minimal word count\n",
        "\n",
        "The following example is the \"Hello, World!\" of data processing, a basic implementation of word count. We're creating a simple data processing pipeline that reads a text file and counts the number of occurrences of every word.\n",
        "\n",
        "There are many scenarios where all the data does not fit in memory. Notice that the outputs of the pipeline go to the file system, which allows for large processing jobs in distributed environments."
      ]
    },
    {
      "metadata": {
        "id": "oUqfqWyMuIfR",
        "outputId": "0b925d3f-72b8-4d4d-d86b-889563594813",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 436
        }
      },
      "cell_type": "code",
      "source": [
        "import apache_beam as beam\n",
        "import re\n",
        "\n",
        "inputs_pattern = 'data/*'\n",
        "outputs_prefix = 'outputs/part'\n",
        "\n",
        "# Running locally in the DirectRunner.\n",
        "with beam.Pipeline() as pipeline:\n",
        "  (\n",
        "      pipeline\n",
        "      | 'Read lines' >> beam.io.ReadFromText(inputs_pattern)\n",
        "      | 'Find words' >> beam.FlatMap(lambda line: re.findall(r\"[a-zA-Z']+\", line))\n",
        "      | 'Pair words with 1' >> beam.Map(lambda word: (word, 1))\n",
        "      | 'Group and sum' >> beam.CombinePerKey(sum)\n",
        "      | 'Format results' >> beam.Map(lambda word_count: str(word_count))\n",
        "      | 'Write results' >> beam.io.WriteToText(outputs_prefix)\n",
        "  )\n",
        "\n",
        "# Sample the first 20 results, remember there are no ordering guarantees.\n",
        "run('head -n 20 {}-00000-of-*'.format(outputs_prefix))"
      ],
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "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"
          ]
        },
        {
          "output_type": "display_data",
          "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": "stream",
          "name": "stdout",
          "text": [
            ">> head -n 20 outputs/part-00000-of-*\n",
            "('KING', 243)\n",
            "('LEAR', 236)\n",
            "('DRAMATIS', 1)\n",
            "('PERSONAE', 1)\n",
            "('king', 65)\n",
            "('of', 447)\n",
            "('Britain', 2)\n",
            "('OF', 15)\n",
            "('FRANCE', 10)\n",
            "('DUKE', 3)\n",
            "('BURGUNDY', 8)\n",
            "('CORNWALL', 63)\n",
            "('ALBANY', 67)\n",
            "('EARL', 2)\n",
            "('KENT', 156)\n",
            "('GLOUCESTER', 141)\n",
            "('EDGAR', 126)\n",
            "('son', 29)\n",
            "('to', 438)\n",
            "('Gloucester', 26)\n",
            "\n"
          ]
        }
      ]
    },
    {
      "metadata": {
        "id": "k-HubCrk-h_G"
      },
      "cell_type": "markdown",
      "source": [
        "# Word count with comments\n",
        "\n",
        "Below is mostly the same code as above, but with comments explaining every line in more detail."
      ]
    },
    {
      "metadata": {
        "id": "x_D7sxUHFzUp",
        "outputId": "44c926df-aa4a-4bea-9247-27c7cb537717",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 1173
        }
      },
      "cell_type": "code",
      "source": [
        "import apache_beam as beam\n",
        "import re\n",
        "\n",
        "inputs_pattern = 'data/*'\n",
        "outputs_prefix = 'outputs/part'\n",
        "\n",
        "# Running locally in the DirectRunner.\n",
        "with beam.Pipeline() as pipeline:\n",
        "  # Store the word counts in a PCollection.\n",
        "  # Each element is a tuple of (word, count) of types (str, int).\n",
        "  word_counts = (\n",
        "      # The input PCollection is an empty pipeline.\n",
        "      pipeline\n",
        "\n",
        "      # Read lines from a text file.\n",
        "      | 'Read lines' >> beam.io.ReadFromText(inputs_pattern)\n",
        "      # Element type: str - text line\n",
        "\n",
        "      # Use a regular expression to iterate over all words in the line.\n",
        "      # FlatMap will yield an element for every element in an iterable.\n",
        "      | 'Find words' >> beam.FlatMap(lambda line: re.findall(r\"[a-zA-Z']+\", line))\n",
        "      # Element type: str - word\n",
        "\n",
        "      # Create key-value pairs where the value is 1, this way we can group by\n",
        "      # the same word while adding those 1s and get the counts for every word.\n",
        "      | 'Pair words with 1' >> beam.Map(lambda word: (word, 1))\n",
        "      # Element type: (str, int) - key: word, value: 1\n",
        "\n",
        "      # Group by key while combining the value using the sum() function.\n",
        "      | 'Group and sum' >> beam.CombinePerKey(sum)\n",
        "      # Element type: (str, int) - key: word, value: counts\n",
        "  )\n",
        "\n",
        "  # We can process a PCollection through other pipelines too.\n",
        "  (\n",
        "      # The input PCollection is the word_counts created from the previous step.\n",
        "      word_counts\n",
        "\n",
        "      # Format the results into a string so we can write them to a file.\n",
        "      | 'Format results' >> beam.Map(lambda word_count: str(word_count))\n",
        "      # Element type: str - text line\n",
        "\n",
        "      # Finally, write the results to a file.\n",
        "      | 'Write results' >> beam.io.WriteToText(outputs_prefix)\n",
        "  )\n",
        "\n",
        "# Sample the first 20 results, remember there are no ordering guarantees.\n",
        "run('head -n 20 {}-00000-of-*'.format(outputs_prefix))"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "WARNING:root:Deleting 1 existing files in target path matching: -*-of-%(num_shards)05d\n"
          ],
          "name": "stderr"
        },
        {
          "output_type": "stream",
          "text": [
            ">> head -n 20 outputs/part-00000-of-*\n",
            "==> outputs/part-00000-of-00001 <==\n",
            "(u'canker', 1)\n",
            "(u'bounty', 2)\n",
            "(u'provision', 3)\n",
            "(u'to', 438)\n",
            "(u'terms', 2)\n",
            "(u'unnecessary', 2)\n",
            "(u'tongue', 5)\n",
            "(u'knives', 1)\n",
            "(u'Commend', 1)\n",
            "(u'Hum', 2)\n",
            "(u'Set', 2)\n",
            "(u'smell', 6)\n",
            "(u'dreadful', 3)\n",
            "(u'frowning', 1)\n",
            "(u'World', 1)\n",
            "(u'tike', 1)\n",
            "(u'yes', 3)\n",
            "(u'oldness', 1)\n",
            "(u'boat', 1)\n",
            "(u\"in's\", 1)\n",
            "\n",
            "==> outputs/part-00000-of-00003 <==\n",
            "wrath: 3\n",
            "nicely: 2\n",
            "hall: 1\n",
            "Sure: 2\n",
            "legs: 4\n",
            "ten: 1\n",
            "yourselves: 1\n",
            "embossed: 1\n",
            "poorly: 1\n",
            "temper: 2\n",
            "Dismissing: 1\n",
            "Legitimate: 1\n",
            "tyrannous: 1\n",
            "turn: 13\n",
            "gold: 2\n",
            "minds: 1\n",
            "dowers: 2\n",
            "policy: 1\n",
            "I: 708\n",
            "V: 6\n",
            "\n",
            "==> outputs/part-00000-of-00004 <==\n",
            "retinue: 1\n",
            "stink: 1\n",
            "beaks: 1\n",
            "Ten: 1\n",
            "riots: 2\n",
            "Service: 1\n",
            "dealing: 1\n",
            "stop: 2\n",
            "detain: 1\n",
            "beware: 1\n",
            "pilferings: 1\n",
            "swimming: 1\n",
            "The: 124\n",
            "Been: 1\n",
            "behavior: 1\n",
            "impetuous: 1\n",
            "Thy: 20\n",
            "Tis: 24\n",
            "Soldiers: 7\n",
            "Juno: 1\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    }
  ]
}