apache_beam_basics_1.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyMTULYiv4kXlCvClq5cZWuE",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/itsmpython/python_for_data_analysis/blob/main/apache_beam_basics_1.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install apache_beam"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "collapsed": true,
        "id": "oLIVtZS4rUgy",
        "outputId": "cdf5c11b-aca1-4240-9176-41833d00d3a5"
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting apache-beam\n",
            "  Downloading apache_beam-2.59.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (7.5 kB)\n",
            "Collecting crcmod<2.0,>=1.7 (from apache-beam)\n",
            "  Downloading crcmod-1.7.tar.gz (89 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m89.7/89.7 kB\u001b[0m \u001b[31m1.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Collecting orjson<4,>=3.9.7 (from apache-beam)\n",
            "  Downloading orjson-3.10.7-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (50 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m50.4/50.4 kB\u001b[0m \u001b[31m437.9 kB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting dill<0.3.2,>=0.3.1.1 (from apache-beam)\n",
            "  Downloading dill-0.3.1.1.tar.gz (151 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m152.0/152.0 kB\u001b[0m \u001b[31m6.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: cloudpickle~=2.2.1 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (2.2.1)\n",
            "Collecting fastavro<2,>=0.23.6 (from apache-beam)\n",
            "  Downloading fastavro-1.9.7-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (5.5 kB)\n",
            "Collecting fasteners<1.0,>=0.3 (from apache-beam)\n",
            "  Downloading fasteners-0.19-py3-none-any.whl.metadata (4.9 kB)\n",
            "Requirement already satisfied: grpcio!=1.48.0,!=1.59.*,!=1.60.*,!=1.61.*,!=1.62.0,!=1.62.1,<2,>=1.33.1 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (1.64.1)\n",
            "Collecting hdfs<3.0.0,>=2.1.0 (from apache-beam)\n",
            "  Downloading hdfs-2.7.3.tar.gz (43 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m43.5/43.5 kB\u001b[0m \u001b[31m2.2 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: httplib2<0.23.0,>=0.8 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (0.22.0)\n",
            "Requirement already satisfied: jsonschema<5.0.0,>=4.0.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (4.23.0)\n",
            "Requirement already satisfied: jsonpickle<4.0.0,>=3.0.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (3.3.0)\n",
            "Requirement already satisfied: numpy<1.27.0,>=1.14.3 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (1.26.4)\n",
            "Collecting objsize<0.8.0,>=0.6.1 (from apache-beam)\n",
            "  Downloading objsize-0.7.0-py3-none-any.whl.metadata (12 kB)\n",
            "Requirement already satisfied: packaging>=22.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (24.1)\n",
            "Collecting pymongo<5.0.0,>=3.8.0 (from apache-beam)\n",
            "  Downloading pymongo-4.8.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (22 kB)\n",
            "Requirement already satisfied: proto-plus<2,>=1.7.1 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (1.24.0)\n",
            "Requirement already satisfied: protobuf!=4.0.*,!=4.21.*,!=4.22.0,!=4.23.*,!=4.24.*,<4.26.0,>=3.20.3 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (3.20.3)\n",
            "Requirement already satisfied: pydot<2,>=1.2.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (1.4.2)\n",
            "Requirement already satisfied: python-dateutil<3,>=2.8.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (2.8.2)\n",
            "Requirement already satisfied: pytz>=2018.3 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (2024.2)\n",
            "Collecting redis<6,>=5.0.0 (from apache-beam)\n",
            "  Downloading redis-5.0.8-py3-none-any.whl.metadata (9.2 kB)\n",
            "Requirement already satisfied: regex>=2020.6.8 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (2024.5.15)\n",
            "Requirement already satisfied: requests<3.0.0,>=2.24.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (2.32.3)\n",
            "Requirement already satisfied: typing-extensions>=3.7.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (4.12.2)\n",
            "Collecting zstandard<1,>=0.18.0 (from apache-beam)\n",
            "  Downloading zstandard-0.23.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (3.0 kB)\n",
            "Requirement already satisfied: pyarrow<17.0.0,>=3.0.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (14.0.2)\n",
            "Requirement already satisfied: pyarrow-hotfix<1 in /usr/local/lib/python3.10/dist-packages (from apache-beam) (0.6)\n",
            "Collecting js2py<1,>=0.74 (from apache-beam)\n",
            "  Downloading Js2Py-0.74-py3-none-any.whl.metadata (868 bytes)\n",
            "Collecting docopt (from hdfs<3.0.0,>=2.1.0->apache-beam)\n",
            "  Downloading docopt-0.6.2.tar.gz (25 kB)\n",
            "  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: six>=1.9.0 in /usr/local/lib/python3.10/dist-packages (from hdfs<3.0.0,>=2.1.0->apache-beam) (1.16.0)\n",
            "Requirement already satisfied: pyparsing!=3.0.0,!=3.0.1,!=3.0.2,!=3.0.3,<4,>=2.4.2 in /usr/local/lib/python3.10/dist-packages (from httplib2<0.23.0,>=0.8->apache-beam) (3.1.4)\n",
            "Requirement already satisfied: tzlocal>=1.2 in /usr/local/lib/python3.10/dist-packages (from js2py<1,>=0.74->apache-beam) (5.2)\n",
            "Collecting pyjsparser>=2.5.1 (from js2py<1,>=0.74->apache-beam)\n",
            "  Downloading pyjsparser-2.7.1.tar.gz (24 kB)\n",
            "  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: attrs>=22.2.0 in /usr/local/lib/python3.10/dist-packages (from jsonschema<5.0.0,>=4.0.0->apache-beam) (24.2.0)\n",
            "Requirement already satisfied: jsonschema-specifications>=2023.03.6 in /usr/local/lib/python3.10/dist-packages (from jsonschema<5.0.0,>=4.0.0->apache-beam) (2023.12.1)\n",
            "Requirement already satisfied: referencing>=0.28.4 in /usr/local/lib/python3.10/dist-packages (from jsonschema<5.0.0,>=4.0.0->apache-beam) (0.35.1)\n",
            "Requirement already satisfied: rpds-py>=0.7.1 in /usr/local/lib/python3.10/dist-packages (from jsonschema<5.0.0,>=4.0.0->apache-beam) (0.20.0)\n",
            "Collecting dnspython<3.0.0,>=1.16.0 (from pymongo<5.0.0,>=3.8.0->apache-beam)\n",
            "  Downloading dnspython-2.6.1-py3-none-any.whl.metadata (5.8 kB)\n",
            "Requirement already satisfied: async-timeout>=4.0.3 in /usr/local/lib/python3.10/dist-packages (from redis<6,>=5.0.0->apache-beam) (4.0.3)\n",
            "Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.24.0->apache-beam) (3.3.2)\n",
            "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.24.0->apache-beam) (3.8)\n",
            "Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.24.0->apache-beam) (2.0.7)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.24.0->apache-beam) (2024.8.30)\n",
            "Downloading apache_beam-2.59.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (15.6 MB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m15.6/15.6 MB\u001b[0m \u001b[31m41.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hDownloading fastavro-1.9.7-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (3.1 MB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m3.1/3.1 MB\u001b[0m \u001b[31m40.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hDownloading fasteners-0.19-py3-none-any.whl (18 kB)\n",
            "Downloading Js2Py-0.74-py3-none-any.whl (1.0 MB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.0/1.0 MB\u001b[0m \u001b[31m32.3 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hDownloading objsize-0.7.0-py3-none-any.whl (11 kB)\n",
            "Downloading orjson-3.10.7-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (141 kB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m141.9/141.9 kB\u001b[0m \u001b[31m8.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hDownloading pymongo-4.8.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (1.2 MB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.2/1.2 MB\u001b[0m \u001b[31m26.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hDownloading redis-5.0.8-py3-none-any.whl (255 kB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m255.6/255.6 kB\u001b[0m \u001b[31m12.9 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hDownloading zstandard-0.23.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (5.4 MB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m5.4/5.4 MB\u001b[0m \u001b[31m56.9 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hDownloading dnspython-2.6.1-py3-none-any.whl (307 kB)\n",
            "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m307.7/307.7 kB\u001b[0m \u001b[31m15.8 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hBuilding wheels for collected packages: crcmod, dill, hdfs, pyjsparser, docopt\n",
            "  Building wheel for crcmod (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for crcmod: filename=crcmod-1.7-cp310-cp310-linux_x86_64.whl size=31408 sha256=11c1573846adf40613c93630bd3d77e9dafe5115b5afffeda5333f6c1cd66bab\n",
            "  Stored in directory: /root/.cache/pip/wheels/85/4c/07/72215c529bd59d67e3dac29711d7aba1b692f543c808ba9e86\n",
            "  Building wheel for dill (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for dill: filename=dill-0.3.1.1-py3-none-any.whl size=78542 sha256=07a4181c139f630efd43148bae069ac91ce2244d9027d849605cec75e3b7b63c\n",
            "  Stored in directory: /root/.cache/pip/wheels/ea/e2/86/64980d90e297e7bf2ce588c2b96e818f5399c515c4bb8a7e4f\n",
            "  Building wheel for hdfs (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for hdfs: filename=hdfs-2.7.3-py3-none-any.whl size=34325 sha256=6604dfd0c293148bbe9fa7899030b028a662d040113f6f8b008ebf0c82f4570a\n",
            "  Stored in directory: /root/.cache/pip/wheels/e5/8d/b6/99c1c0a3ac5788c866b0ecd3f48b0134a5910e6ed26011800b\n",
            "  Building wheel for pyjsparser (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyjsparser: filename=pyjsparser-2.7.1-py3-none-any.whl size=25983 sha256=e2d7fdf0ac32e000f35606177b3f4cd0d8f2f328fad59ee0073cd4f1351f63fa\n",
            "  Stored in directory: /root/.cache/pip/wheels/5e/81/26/5956478df303e2bf5a85a5df595bb307bd25948a4bab69f7c7\n",
            "  Building wheel for docopt (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for docopt: filename=docopt-0.6.2-py2.py3-none-any.whl size=13704 sha256=35ec0db869dc59d861961a07b480ff8172e5c4c09cb584c37cdd92c2f2565c8a\n",
            "  Stored in directory: /root/.cache/pip/wheels/fc/ab/d4/5da2067ac95b36618c629a5f93f809425700506f72c9732fac\n",
            "Successfully built crcmod dill hdfs pyjsparser docopt\n",
            "Installing collected packages: pyjsparser, docopt, crcmod, zstandard, redis, orjson, objsize, js2py, fasteners, fastavro, dnspython, dill, pymongo, hdfs, apache-beam\n",
            "Successfully installed apache-beam-2.59.0 crcmod-1.7 dill-0.3.1.1 dnspython-2.6.1 docopt-0.6.2 fastavro-1.9.7 fasteners-0.19 hdfs-2.7.3 js2py-0.74 objsize-0.7.0 orjson-3.10.7 pyjsparser-2.7.1 pymongo-4.8.0 redis-5.0.8 zstandard-0.23.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import apache_beam as beam"
      ],
      "metadata": {
        "id": "ENrBsHihrkiQ"
      },
      "execution_count": 2,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "pipe = beam.Pipeline() # Define your pipline object"
      ],
      "metadata": {
        "id": "h1HE8y2Xro7n"
      },
      "execution_count": 22,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Uploaded the file mtcars.csv to Google Colab and reading the file now\n",
        "ip =(\n",
        "    pipe\n",
        "    | beam.io.ReadFromText('/content/mtcars.csv', skip_header_lines=1) # |symbol is apply function\n",
        "    | beam.Map(lambda record: record.split(',')) # Since we read csv as txt, let us split it using ',' which will create list of lists. Each row is a list\n",
        "    | beam.Filter(lambda record: record[10] == '4') # Apply a filter function\n",
        "    #| beam.io.WriteToText('output.txt') # write to a text file\n",
        "    | beam.combiners.Count.Globally()\n",
        "    | beam.Map(print) # Print the output\n",
        ")"
      ],
      "metadata": {
        "id": "L2QSQEdXtKfo"
      },
      "execution_count": 24,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "pipe.run()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "i0eb3OEfu2d6",
        "outputId": "11352fd1-0400-402f-90f6-2375ff5d4ce6"
      },
      "execution_count": 25,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "12\n",
            "12\n"
          ]
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "<apache_beam.runners.portability.fn_api_runner.fn_runner.RunnerResult at 0x7d35f3c3fcd0>"
            ]
          },
          "metadata": {},
          "execution_count": 25
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Instead of initializing the Pipe everytime and then running it, we can use the following method\n",
        "![green-divider](https://user-images.githubusercontent.com/7065401/52071924-c003ad80-2562-11e9-8297-1c6595f8a7ff.png)\n"
      ],
      "metadata": {
        "id": "ufBL2_5PyKdH"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "with beam.Pipeline() as pipe1:\n",
        "  ip1 =(\n",
        "    pipe1\n",
        "    | beam.io.ReadFromText('/content/mtcars.csv', skip_header_lines=1) # |symbol is apply function\n",
        "    | beam.Map(lambda record: record.split(',')) # Since we read csv as txt, let us split it using ',' which will create list of lists. Each row is a list\n",
        "    | beam.Filter(lambda record: record[10] == '4') # Apply a filter function\n",
        "    #| beam.io.WriteToText('output.txt') # write to a text file\n",
        "    | beam.combiners.Count.Globally() # Globally, becuase out data would be processed on distributed computing\n",
        "    | beam.Map(print) # Print the output\n",
        ")"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "sfcdNH9MyWQD",
        "outputId": "db50bef9-afa7-4b77-8456-6c9b9fe5c750"
      },
      "execution_count": 26,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "12\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Using custom functions\n",
        "![green-divider](https://user-images.githubusercontent.com/7065401/52071924-c003ad80-2562-11e9-8297-1c6595f8a7ff.png)\n",
        "\n",
        "We are using lambda function for simple processing. What if we have complex processes?\n",
        "- We can use custom functions for that. Let us see how"
      ],
      "metadata": {
        "id": "BZcWBZZJ1_zZ"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def filter_data(element):\n",
        "  return element[10] == '4'\n",
        "#"
      ],
      "metadata": {
        "id": "dLuUmXXr2NrG"
      },
      "execution_count": 55,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "with beam.Pipeline() as pipe2:\n",
        "  ip2 =(\n",
        "    pipe2\n",
        "    | beam.io.ReadFromText('/content/mtcars.csv', skip_header_lines=1) # |symbol is apply function\n",
        "    | beam.Map(lambda record: record.split(',')) # Since we read csv as txt, let us split it using ',' which will create list of lists. Each row is a list\n",
        "    | beam.Filter(filter_data) # Apply a filter function\n",
        "    | beam.combiners.Count.Globally() # Globally, becuase out data would be processed on distributed computing\n",
        "    #| beam.Map(print) # Print the output\n",
        "    | beam.io.WriteToText('output.txt') # write to a text file\n",
        ")"
      ],
      "metadata": {
        "id": "IzPj7C3k2V2D"
      },
      "execution_count": 56,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Aggregations in Apache beam using *PerKey()*\n",
        "Apply an aggregator function which acts like groupby\n",
        "\n",
        "![green-divider](https://user-images.githubusercontent.com/7065401/52071924-c003ad80-2562-11e9-8297-1c6595f8a7ff.png)\n",
        "\n"
      ],
      "metadata": {
        "id": "oyUP87x3BxNu"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "with beam.Pipeline() as pipe3:\n",
        "  ip3 =(\n",
        "    pipe3\n",
        "    | beam.io.ReadFromText('/content/mtcars.csv', skip_header_lines=1) # |symbol is apply function\n",
        "    | beam.Map(lambda record: record.split(',')) # Since we read csv as txt, let us split it using ',' which will create list of lists. Each row is a list\n",
        "    | beam.Map(lambda record:(record[2], record)) # Apply an aggregator function which acts like\n",
        "    | beam.combiners.Count.PerKey() # Count PerKey to get the aggregation\n",
        "    | beam.Map(print) # Print the output\n",
        "    #| beam.io.WriteToText('output.txt') # write to a text file\n",
        ")"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "IIdviqcVB7W_",
        "outputId": "a4896314-d869-4f64-897f-4fed1eee190b"
      },
      "execution_count": 57,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "('6', 7)\n",
            "('4', 11)\n",
            "('8', 14)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Creating and processing *local/inmemory pCollection*\n",
        "![green-divider](https://user-images.githubusercontent.com/7065401/52071924-c003ad80-2562-11e9-8297-1c6595f8a7ff.png)\n",
        "\n"
      ],
      "metadata": {
        "id": "RtKVeCZQEffE"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "a=[12,11,10,9,8,7,6,5,4,3,2,1,0]\n",
        "with beam.Pipeline() as pipe1:\n",
        "  ip1 =(pipe1\n",
        "        |beam.Create(a)\n",
        "        |beam.Filter(lambda x: x%2 != 0)\n",
        "        |beam.Map(print)\n",
        "  )"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Rn2rw1VtEvmq",
        "outputId": "0499701f-4b3a-4944-9a7f-51ce8ace3e45"
      },
      "execution_count": 50,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "11\n",
            "9\n",
            "7\n",
            "5\n",
            "3\n",
            "1\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Aggregation using multiple keys using *PerKey()*\n",
        "Apply an aggregator function on multiple keys\n",
        "\n",
        "![green-divider](https://user-images.githubusercontent.com/7065401/52071924-c003ad80-2562-11e9-8297-1c6595f8a7ff.png)\n",
        "\n"
      ],
      "metadata": {
        "id": "ZAnAg9sBGSVE"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "with beam.Pipeline() as pipe4:\n",
        "  # Read data from CSV, split each line by comma\n",
        "  data = (\n",
        "      pipe4\n",
        "      | 'ReadFromText' >> beam.io.ReadFromText('/content/mtcars.csv', skip_header_lines=1)\n",
        "      | 'SplitLines' >> beam.Map(lambda record: record.split(','))\n",
        "  )\n",
        "\n",
        "  # Extract model (Model 0) and mpg (index 10) as key-value pairs\n",
        "  # You can modify the indices based on your actual column positions\n",
        "  model_mpg = (\n",
        "      data\n",
        "      | 'ExtractModelMPG' >> beam.Map(lambda record: (record[2], float(record[1])))  # Assuming mpg is numeric\n",
        "  )\n",
        "\n",
        "  # Count occurrences of each model-mpg combination\n",
        "  # This will output (model, mpg), count\n",
        "  model_mpg_counts = model_mpg | 'CountPerModelMPG' >> beam.combiners.Count.PerKey()\n",
        "  #let us print the results\n",
        "  model_mpg_counts | 'PrintResults' >> beam.Map(print)\n",
        "\n",
        "  # Optionally write results to a text file\n",
        "  #model_mpg_counts | 'WriteToText' >> beam.io.WriteToText('output.txt')\n",
        "\n",
        "  # You can further process model_mpg_counts here (e.g., filter, sort)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "IZJMUTbhcNw6",
        "outputId": "5f81def2-240f-4375-9ed2-3489b9d0f14f"
      },
      "execution_count": 77,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "('6', 7)\n",
            "('4', 11)\n",
            "('8', 14)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Following is an *independent function* used in a pipeline\n",
        "  This splits the package names and returns a list\n",
        "  ![green-divider](https://user-images.githubusercontent.com/7065401/52071924-c003ad80-2562-11e9-8297-1c6595f8a7ff.png)\n"
      ],
      "metadata": {
        "id": "-JXmbJwGyyki"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "id": "h3iYRzqydK3b"
      },
      "outputs": [],
      "source": [
        "def splitPackageName(packageName):\n",
        "   \"\"\"e.g. given com.example.appname.library.widgetname\n",
        "           returns com\n",
        "\t           com.example\n",
        "                   com.example.appname\n",
        "      etc.\n",
        "   \"\"\"\n",
        "   result = []\n",
        "   end = packageName.find('.')\n",
        "   while end > 0:\n",
        "      result.append(packageName[0:end])\n",
        "      end = packageName.find('.', end+1)\n",
        "   # result.append(packageName) # To capture the final package name\n",
        "   return result"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "re = splitPackageName('com.example.appname.library.widgetname')\n",
        "re\n",
        "\n",
        "# ['com',\n",
        "#  'com.example',\n",
        "#  'com.example.appname',\n",
        "#  'com.example.appname.library',\n",
        "#  'com.example.appname.library.widgetname']"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "LoVEdtridRdW",
        "outputId": "6f18e626-f6d3-4eaf-e14b-6d3ad285ac62"
      },
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "['com',\n",
              " 'com.example',\n",
              " 'com.example.appname',\n",
              " 'com.example.appname.library',\n",
              " 'com.example.appname.library.widgetname']"
            ]
          },
          "metadata": {},
          "execution_count": 7
        }
      ]
    }
  ]
}