MapReduce_Primer_HelloWorld.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "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/groda/big_data/blob/master/MapReduce_Primer_HelloWorld.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "<a href=\"https://github.com/groda/big_data\"><div><img src=\"https://github.com/groda/big_data/blob/master/logo_bdb.png?raw=true\" align=right width=\"90\"></div></a>\n",
        "\n",
        "# MapReduce: A Primer with <code>Hello World!</code>\n",
        "<br>\n",
        "<br>\n",
        "\n",
        "For this tutorial, we are going to download the core Hadoop distribution and run Hadoop in _local standalone mode_:\n",
        "\n",
        "> ❝ _By default, Hadoop is configured to run in a non-distributed mode, as a single Java process._ ❞\n",
        "\n",
        "(see [https://hadoop.apache.org/docs/stable/.../Standalone_Operation](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/SingleCluster.html#Standalone_Operation))\n",
        "\n",
        "We are going to run a MapReduce job using MapReduce's [streaming application](https://hadoop.apache.org/docs/stable/hadoop-streaming/HadoopStreaming.html#Hadoop_Streaming). This is not to be confused with real-time streaming:\n",
        "\n",
        "> ❝ _Hadoop streaming is a utility that comes with the Hadoop distribution. The utility allows you to create and run Map/Reduce jobs with any executable or script as the mapper and/or the reducer._ ❞\n",
        "\n",
        "MapReduce streaming defaults to using [`IdentityMapper`](https://hadoop.apache.org/docs/stable/api/index.html) and [`IdentityReducer`](https://hadoop.apache.org/docs/stable/api/index.html), thus eliminating the need for explicit specification of a mapper or reducer. Finally, we show how to run a map-only job by setting `mapreduce.job.reduce` equal to $0$.\n",
        "\n",
        "Both input and output are standard files since Hadoop's default filesystem is the regular file system, as specified by the `fs.defaultFS` property in [core-default.xml](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/core-default.xml)).\n"
      ],
      "metadata": {
        "id": "GzbmlR27wh6e"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Download core Hadoop"
      ],
      "metadata": {
        "id": "uUbM5R0GwwYw"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "HADOOP_URL = \"https://dlcdn.apache.org/hadoop/common/stable/hadoop-3.4.0.tar.gz\"\n",
        "\n",
        "import requests\n",
        "import os\n",
        "import tarfile\n",
        "\n",
        "def download_and_extract_targz(url):\n",
        "    response = requests.get(url)\n",
        "    filename = url.rsplit('/', 1)[-1]\n",
        "    HADOOP_HOME = filename[:-7]\n",
        "    # set HADOOP_HOME environment variable\n",
        "    os.environ['HADOOP_HOME'] = HADOOP_HOME\n",
        "    if os.path.isdir(HADOOP_HOME):\n",
        "      print(\"Not downloading, Hadoop folder {} already exists\".format(HADOOP_HOME))\n",
        "      return\n",
        "    if response.status_code == 200:\n",
        "        with open(filename, 'wb') as file:\n",
        "            file.write(response.content)\n",
        "        with tarfile.open(filename, 'r:gz') as tar_ref:\n",
        "            extract_path = tar_ref.extractall(path='.')\n",
        "            # Get the names of all members (files and directories) in the archive\n",
        "            all_members = tar_ref.getnames()\n",
        "            # If there is a top-level directory, get its name\n",
        "            if all_members:\n",
        "              top_level_directory = all_members[0]\n",
        "              print(f\"ZIP file downloaded and extracted successfully. Contents saved at: {top_level_directory}\")\n",
        "    else:\n",
        "        print(f\"Failed to download ZIP file. Status code: {response.status_code}\")\n",
        "\n",
        "\n",
        "download_and_extract_targz(HADOOP_URL)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "jDgQtQlzw8bL",
        "outputId": "1e21fc75-4964-4181-ec72-e224e0493c54"
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "ZIP file downloaded and extracted successfully. Contents saved at: hadoop-3.4.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Set environment variables"
      ],
      "metadata": {
        "id": "3yvb5cw9xEbh"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Set `HADOOP_HOME` and `PATH`"
      ],
      "metadata": {
        "id": "u6lkrz1dxIiO"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# HADOOP_HOME was set earlier when downloading Hadoop distribution\n",
        "print(\"HADOOP_HOME is {}\".format(os.environ['HADOOP_HOME']))\n",
        "\n",
        "os.environ['PATH'] = ':'.join([os.path.join(os.environ['HADOOP_HOME'], 'bin'), os.environ['PATH']])\n",
        "print(\"PATH is {}\".format(os.environ['PATH']))"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "s7maAwaFxBT_",
        "outputId": "6165ab81-b309-4e14-bb0d-134bdf3970cf"
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "HADOOP_HOME is hadoop-3.4.0\n",
            "PATH is hadoop-3.4.0/bin:/opt/bin:/usr/local/nvidia/bin:/usr/local/cuda/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/tools/node/bin:/tools/google-cloud-sdk/bin\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Set `JAVA_HOME`\n",
        "\n",
        "While Java is readily available on Google Colab, we consider the broader scenario of an Ubuntu machine. In this case, we ensure compatibility by installing Java, specifically opting for the `openjdk-19-jre-headless` version."
      ],
      "metadata": {
        "id": "4kzJ8cNoxPyK"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import shutil\n",
        "\n",
        "# set variable JAVA_HOME (install Java if necessary)\n",
        "def is_java_installed():\n",
        "    os.environ['JAVA_HOME'] = os.path.realpath(shutil.which(\"java\")).split('/bin')[0]\n",
        "    return os.environ['JAVA_HOME']\n",
        "\n",
        "def install_java():\n",
        "    # Uncomment and modify the desired version\n",
        "    # java_version= 'openjdk-11-jre-headless'\n",
        "    # java_version= 'default-jre'\n",
        "    # java_version= 'openjdk-17-jre-headless'\n",
        "    # java_version= 'openjdk-18-jre-headless'\n",
        "    java_version= 'openjdk-19-jre-headless'\n",
        "\n",
        "    print(f\"Java not found. Installing {java_version} ... (this might take a while)\")\n",
        "    try:\n",
        "        cmd = f\"apt install -y {java_version}\"\n",
        "        subprocess_output = subprocess.run(cmd, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)\n",
        "        stdout_result = subprocess_output.stdout\n",
        "        # Process the results as needed\n",
        "        print(\"Done installing Java {}\".format(java_version))\n",
        "        os.environ['JAVA_HOME'] = os.path.realpath(shutil.which(\"java\")).split('/bin')[0]\n",
        "        print(\"JAVA_HOME is {}\".format(os.environ['JAVA_HOME']))\n",
        "    except subprocess.CalledProcessError as e:\n",
        "        # Handle the error if the command returns a non-zero exit code\n",
        "        print(\"Command failed with return code {}\".format(e.returncode))\n",
        "        print(\"stdout: {}\".format(e.stdout))\n",
        "\n",
        "# Install Java if not available\n",
        "if is_java_installed():\n",
        "    print(\"Java is already installed: {}\".format(os.environ['JAVA_HOME']))\n",
        "else:\n",
        "    print(\"Installing Java\")\n",
        "    install_java()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "SauFHVPOxL-Y",
        "outputId": "2dfc8233-4428-4ba1-e73e-c6d8f318dad8"
      },
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Java is already installed: /usr/lib/jvm/java-11-openjdk-amd64\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Run a MapReduce job with Hadoop streaming"
      ],
      "metadata": {
        "id": "6HFPVX84xbNd"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Create a file\n",
        "\n",
        "Write the string\"Hello, World!\" to a local file.<p>**Note:** you will be writing to the file `./hello.txt` in your current directory (denoted by `./`)."
      ],
      "metadata": {
        "id": "_yVa55X1xmOb"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!echo \"Hello, World!\">./hello.txt"
      ],
      "metadata": {
        "id": "9Jz7mJkcxYxw"
      },
      "execution_count": 4,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Launch the MapReduce \"Hello, World!\" application\n",
        "\n",
        "Since the default filesystem is the local filesystem (as opposed to HDFS) we do not need to upload the local file `hello.txt` to HDFS.\n",
        "\n",
        "Run a MapReduce job with `/bin/cat` as a mapper and no reducer.\n",
        "\n",
        "**Note:** the first step of removing the output directory is necessary because MapReduce does not overwrite data folders by design."
      ],
      "metadata": {
        "id": "zSh_Kr5Bxvst"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%bash\n",
        "hdfs dfs -rm -r my_output\n",
        "\n",
        "mapred streaming \\\n",
        "    -input hello.txt \\\n",
        "    -output my_output \\\n",
        "    -mapper '/bin/cat'"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "nb5JryK9xpPA",
        "outputId": "fc063d86-84ac-40b1-defb-6073fcf4759b"
      },
      "execution_count": 5,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "rm: `my_output': No such file or directory\n",
            "2024-03-24 19:44:18,184 INFO impl.MetricsConfig: Loaded properties from hadoop-metrics2.properties\n",
            "2024-03-24 19:44:18,515 INFO impl.MetricsSystemImpl: Scheduled Metric snapshot period at 10 second(s).\n",
            "2024-03-24 19:44:18,515 INFO impl.MetricsSystemImpl: JobTracker metrics system started\n",
            "2024-03-24 19:44:18,554 WARN impl.MetricsSystemImpl: JobTracker metrics system already initialized!\n",
            "2024-03-24 19:44:18,982 INFO mapred.FileInputFormat: Total input files to process : 1\n",
            "2024-03-24 19:44:19,037 INFO mapreduce.JobSubmitter: number of splits:1\n",
            "2024-03-24 19:44:19,515 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_local402428322_0001\n",
            "2024-03-24 19:44:19,515 INFO mapreduce.JobSubmitter: Executing with tokens: []\n",
            "2024-03-24 19:44:19,830 INFO mapreduce.Job: The url to track the job: http://localhost:8080/\n",
            "2024-03-24 19:44:19,833 INFO mapreduce.Job: Running job: job_local402428322_0001\n",
            "2024-03-24 19:44:19,849 INFO mapred.LocalJobRunner: OutputCommitter set in config null\n",
            "2024-03-24 19:44:19,854 INFO mapred.LocalJobRunner: OutputCommitter is org.apache.hadoop.mapred.FileOutputCommitter\n",
            "2024-03-24 19:44:19,875 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:19,875 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:19,989 INFO mapred.LocalJobRunner: Waiting for map tasks\n",
            "2024-03-24 19:44:20,002 INFO mapred.LocalJobRunner: Starting task: attempt_local402428322_0001_m_000000_0\n",
            "2024-03-24 19:44:20,068 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:20,069 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:20,124 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]\n",
            "2024-03-24 19:44:20,139 INFO mapred.MapTask: Processing split: file:/content/hello.txt:0+14\n",
            "2024-03-24 19:44:20,158 INFO mapred.MapTask: numReduceTasks: 1\n",
            "2024-03-24 19:44:20,261 INFO mapred.MapTask: (EQUATOR) 0 kvi 26214396(104857584)\n",
            "2024-03-24 19:44:20,261 INFO mapred.MapTask: mapreduce.task.io.sort.mb: 100\n",
            "2024-03-24 19:44:20,261 INFO mapred.MapTask: soft limit at 83886080\n",
            "2024-03-24 19:44:20,261 INFO mapred.MapTask: bufstart = 0; bufvoid = 104857600\n",
            "2024-03-24 19:44:20,261 INFO mapred.MapTask: kvstart = 26214396; length = 6553600\n",
            "2024-03-24 19:44:20,267 INFO mapred.MapTask: Map output collector class = org.apache.hadoop.mapred.MapTask$MapOutputBuffer\n",
            "2024-03-24 19:44:20,282 INFO streaming.PipeMapRed: PipeMapRed exec [/bin/cat]\n",
            "2024-03-24 19:44:20,294 INFO Configuration.deprecation: mapred.work.output.dir is deprecated. Instead, use mapreduce.task.output.dir\n",
            "2024-03-24 19:44:20,297 INFO Configuration.deprecation: mapred.local.dir is deprecated. Instead, use mapreduce.cluster.local.dir\n",
            "2024-03-24 19:44:20,298 INFO Configuration.deprecation: map.input.file is deprecated. Instead, use mapreduce.map.input.file\n",
            "2024-03-24 19:44:20,298 INFO Configuration.deprecation: map.input.length is deprecated. Instead, use mapreduce.map.input.length\n",
            "2024-03-24 19:44:20,299 INFO Configuration.deprecation: mapred.job.id is deprecated. Instead, use mapreduce.job.id\n",
            "2024-03-24 19:44:20,299 INFO Configuration.deprecation: mapred.task.partition is deprecated. Instead, use mapreduce.task.partition\n",
            "2024-03-24 19:44:20,301 INFO Configuration.deprecation: map.input.start is deprecated. Instead, use mapreduce.map.input.start\n",
            "2024-03-24 19:44:20,302 INFO Configuration.deprecation: mapred.task.is.map is deprecated. Instead, use mapreduce.task.ismap\n",
            "2024-03-24 19:44:20,302 INFO Configuration.deprecation: mapred.task.id is deprecated. Instead, use mapreduce.task.attempt.id\n",
            "2024-03-24 19:44:20,303 INFO Configuration.deprecation: mapred.tip.id is deprecated. Instead, use mapreduce.task.id\n",
            "2024-03-24 19:44:20,304 INFO Configuration.deprecation: mapred.skip.on is deprecated. Instead, use mapreduce.job.skiprecords\n",
            "2024-03-24 19:44:20,304 INFO Configuration.deprecation: user.name is deprecated. Instead, use mapreduce.job.user.name\n",
            "2024-03-24 19:44:20,330 INFO streaming.PipeMapRed: R/W/S=1/0/0 in:NA [rec/s] out:NA [rec/s]\n",
            "2024-03-24 19:44:20,335 INFO streaming.PipeMapRed: Records R/W=1/1\n",
            "2024-03-24 19:44:20,335 INFO streaming.PipeMapRed: MRErrorThread done\n",
            "2024-03-24 19:44:20,338 INFO streaming.PipeMapRed: mapRedFinished\n",
            "2024-03-24 19:44:20,343 INFO mapred.LocalJobRunner: \n",
            "2024-03-24 19:44:20,344 INFO mapred.MapTask: Starting flush of map output\n",
            "2024-03-24 19:44:20,344 INFO mapred.MapTask: Spilling map output\n",
            "2024-03-24 19:44:20,344 INFO mapred.MapTask: bufstart = 0; bufend = 15; bufvoid = 104857600\n",
            "2024-03-24 19:44:20,344 INFO mapred.MapTask: kvstart = 26214396(104857584); kvend = 26214396(104857584); length = 1/6553600\n",
            "2024-03-24 19:44:20,357 INFO mapred.MapTask: Finished spill 0\n",
            "2024-03-24 19:44:20,383 INFO mapred.Task: Task:attempt_local402428322_0001_m_000000_0 is done. And is in the process of committing\n",
            "2024-03-24 19:44:20,388 INFO mapred.LocalJobRunner: Records R/W=1/1\n",
            "2024-03-24 19:44:20,388 INFO mapred.Task: Task 'attempt_local402428322_0001_m_000000_0' done.\n",
            "2024-03-24 19:44:20,408 INFO mapred.Task: Final Counters for attempt_local402428322_0001_m_000000_0: Counters: 17\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=141914\n",
            "\t\tFILE: Number of bytes written=854181\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tMap output bytes=15\n",
            "\t\tMap output materialized bytes=23\n",
            "\t\tInput split bytes=75\n",
            "\t\tCombine input records=0\n",
            "\t\tSpilled Records=1\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=0\n",
            "\t\tGC time elapsed (ms)=15\n",
            "\t\tTotal committed heap usage (bytes)=339738624\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "2024-03-24 19:44:20,409 INFO mapred.LocalJobRunner: Finishing task: attempt_local402428322_0001_m_000000_0\n",
            "2024-03-24 19:44:20,410 INFO mapred.LocalJobRunner: map task executor complete.\n",
            "2024-03-24 19:44:20,414 INFO mapred.LocalJobRunner: Waiting for reduce tasks\n",
            "2024-03-24 19:44:20,415 INFO mapred.LocalJobRunner: Starting task: attempt_local402428322_0001_r_000000_0\n",
            "2024-03-24 19:44:20,433 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:20,433 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:20,434 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]\n",
            "2024-03-24 19:44:20,443 INFO mapred.ReduceTask: Using ShuffleConsumerPlugin: org.apache.hadoop.mapreduce.task.reduce.Shuffle@1bd861ad\n",
            "2024-03-24 19:44:20,446 WARN impl.MetricsSystemImpl: JobTracker metrics system already initialized!\n",
            "2024-03-24 19:44:20,477 INFO reduce.MergeManagerImpl: MergerManager: memoryLimit=2382574336, maxSingleShuffleLimit=595643584, mergeThreshold=1572499072, ioSortFactor=10, memToMemMergeOutputsThreshold=10\n",
            "2024-03-24 19:44:20,480 INFO reduce.EventFetcher: attempt_local402428322_0001_r_000000_0 Thread started: EventFetcher for fetching Map Completion Events\n",
            "2024-03-24 19:44:20,545 INFO reduce.LocalFetcher: localfetcher#1 about to shuffle output of map attempt_local402428322_0001_m_000000_0 decomp: 19 len: 23 to MEMORY\n",
            "2024-03-24 19:44:20,555 INFO reduce.InMemoryMapOutput: Read 19 bytes from map-output for attempt_local402428322_0001_m_000000_0\n",
            "2024-03-24 19:44:20,559 INFO reduce.MergeManagerImpl: closeInMemoryFile -> map-output of size: 19, inMemoryMapOutputs.size() -> 1, commitMemory -> 0, usedMemory ->19\n",
            "2024-03-24 19:44:20,563 INFO reduce.EventFetcher: EventFetcher is interrupted.. Returning\n",
            "2024-03-24 19:44:20,565 INFO mapred.LocalJobRunner: 1 / 1 copied.\n",
            "2024-03-24 19:44:20,565 INFO reduce.MergeManagerImpl: finalMerge called with 1 in-memory map-outputs and 0 on-disk map-outputs\n",
            "2024-03-24 19:44:20,579 INFO mapred.Merger: Merging 1 sorted segments\n",
            "2024-03-24 19:44:20,579 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 3 bytes\n",
            "2024-03-24 19:44:20,581 INFO reduce.MergeManagerImpl: Merged 1 segments, 19 bytes to disk to satisfy reduce memory limit\n",
            "2024-03-24 19:44:20,582 INFO reduce.MergeManagerImpl: Merging 1 files, 23 bytes from disk\n",
            "2024-03-24 19:44:20,583 INFO reduce.MergeManagerImpl: Merging 0 segments, 0 bytes from memory into reduce\n",
            "2024-03-24 19:44:20,584 INFO mapred.Merger: Merging 1 sorted segments\n",
            "2024-03-24 19:44:20,585 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 3 bytes\n",
            "2024-03-24 19:44:20,586 INFO mapred.LocalJobRunner: 1 / 1 copied.\n",
            "2024-03-24 19:44:20,599 INFO mapred.Task: Task:attempt_local402428322_0001_r_000000_0 is done. And is in the process of committing\n",
            "2024-03-24 19:44:20,603 INFO mapred.LocalJobRunner: 1 / 1 copied.\n",
            "2024-03-24 19:44:20,604 INFO mapred.Task: Task attempt_local402428322_0001_r_000000_0 is allowed to commit now\n",
            "2024-03-24 19:44:20,613 INFO output.FileOutputCommitter: Saved output of task 'attempt_local402428322_0001_r_000000_0' to file:/content/my_output\n",
            "2024-03-24 19:44:20,615 INFO mapred.LocalJobRunner: reduce > reduce\n",
            "2024-03-24 19:44:20,615 INFO mapred.Task: Task 'attempt_local402428322_0001_r_000000_0' done.\n",
            "2024-03-24 19:44:20,616 INFO mapred.Task: Final Counters for attempt_local402428322_0001_r_000000_0: Counters: 24\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=141992\n",
            "\t\tFILE: Number of bytes written=854231\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tCombine input records=0\n",
            "\t\tCombine output records=0\n",
            "\t\tReduce input groups=1\n",
            "\t\tReduce shuffle bytes=23\n",
            "\t\tReduce input records=1\n",
            "\t\tReduce output records=1\n",
            "\t\tSpilled Records=1\n",
            "\t\tShuffled Maps =1\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=1\n",
            "\t\tGC time elapsed (ms)=0\n",
            "\t\tTotal committed heap usage (bytes)=339738624\n",
            "\tShuffle Errors\n",
            "\t\tBAD_ID=0\n",
            "\t\tCONNECTION=0\n",
            "\t\tIO_ERROR=0\n",
            "\t\tWRONG_LENGTH=0\n",
            "\t\tWRONG_MAP=0\n",
            "\t\tWRONG_REDUCE=0\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=27\n",
            "2024-03-24 19:44:20,616 INFO mapred.LocalJobRunner: Finishing task: attempt_local402428322_0001_r_000000_0\n",
            "2024-03-24 19:44:20,617 INFO mapred.LocalJobRunner: reduce task executor complete.\n",
            "2024-03-24 19:44:20,853 INFO mapreduce.Job: Job job_local402428322_0001 running in uber mode : false\n",
            "2024-03-24 19:44:20,855 INFO mapreduce.Job:  map 100% reduce 100%\n",
            "2024-03-24 19:44:20,856 INFO mapreduce.Job: Job job_local402428322_0001 completed successfully\n",
            "2024-03-24 19:44:20,867 INFO mapreduce.Job: Counters: 30\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=283906\n",
            "\t\tFILE: Number of bytes written=1708412\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tMap output bytes=15\n",
            "\t\tMap output materialized bytes=23\n",
            "\t\tInput split bytes=75\n",
            "\t\tCombine input records=0\n",
            "\t\tCombine output records=0\n",
            "\t\tReduce input groups=1\n",
            "\t\tReduce shuffle bytes=23\n",
            "\t\tReduce input records=1\n",
            "\t\tReduce output records=1\n",
            "\t\tSpilled Records=2\n",
            "\t\tShuffled Maps =1\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=1\n",
            "\t\tGC time elapsed (ms)=15\n",
            "\t\tTotal committed heap usage (bytes)=679477248\n",
            "\tShuffle Errors\n",
            "\t\tBAD_ID=0\n",
            "\t\tCONNECTION=0\n",
            "\t\tIO_ERROR=0\n",
            "\t\tWRONG_LENGTH=0\n",
            "\t\tWRONG_MAP=0\n",
            "\t\tWRONG_REDUCE=0\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=27\n",
            "2024-03-24 19:44:20,867 INFO streaming.StreamJob: Output directory: my_output\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Verify the result\n",
        "\n",
        "If the job executed successfully, an empty file named `_SUCCESS` is expected to be present in the output directory `my_output`.\n",
        "\n",
        "Verify the success of the MapReduce job by checking for the presence of the `_SUCCESS` file."
      ],
      "metadata": {
        "id": "OB_fX9u5x55y"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%bash\n",
        "\n",
        "echo \"Check if MapReduce job was successful\"\n",
        "hdfs dfs -test -e my_output/_SUCCESS\n",
        "if [ $? -eq 0 ]; then\n",
        "\techo \"_SUCCESS exists!\"\n",
        "fi"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "bnvEvYDfx2g4",
        "outputId": "7025fe37-737a-433f-a865-0380d9891a48"
      },
      "execution_count": 6,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Check if MapReduce job was successful\n",
            "_SUCCESS exists!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "**Note:** `hdfs dfs -ls` is the same as `ls` since the default filesystem is the local filesystem."
      ],
      "metadata": {
        "id": "BLMnBh44x_YR"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!hdfs dfs -ls my_output"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ufAfmGUvx8jW",
        "outputId": "1bff0e0e-5d19-4fa3-b183-621b7df56cdd"
      },
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Found 2 items\n",
            "-rw-r--r--   1 root root          0 2024-03-24 19:44 my_output/_SUCCESS\n",
            "-rw-r--r--   1 root root         15 2024-03-24 19:44 my_output/part-00000\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!ls -l my_output"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ZnKSahPzyCAn",
        "outputId": "176264a3-cf17-4515-fa44-1ac177a4571f"
      },
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "total 4\n",
            "-rw-r--r-- 1 root root 15 Mar 24 19:44 part-00000\n",
            "-rw-r--r-- 1 root root  0 Mar 24 19:44 _SUCCESS\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "The actual output of the MapReduce job is contained in the file `part-00000` in the output directory."
      ],
      "metadata": {
        "id": "v9LmpcaMyG23"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!cat my_output/part-00000"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "eL-Clat5yD8I",
        "outputId": "8add6017-333f-429d-caff-e824282714ff"
      },
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Hello, World!\t\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# MapReduce without specifying mapper or reducer\n",
        "\n",
        "In the previous example, we have seen how to run a MapReduce job without specifying any reducer.\n",
        "\n",
        "Since the only required options for `mapred streaming` are `input` and `output`, we can also run a MapReduce job without specifying a mapper."
      ],
      "metadata": {
        "id": "AmpHr_HyyMnM"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!mapred streaming -h"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ZPWL1AiXyJac",
        "outputId": "450786f5-701d-42d0-ecd1-986ac5317e0a"
      },
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "2024-03-24 19:44:27,372 ERROR streaming.StreamJob: Unrecognized option: -h\n",
            "Usage: $HADOOP_HOME/bin/hadoop jar hadoop-streaming.jar [options]\n",
            "Options:\n",
            "  -input          <path> DFS input file(s) for the Map step.\n",
            "  -output         <path> DFS output directory for the Reduce step.\n",
            "  -mapper         <cmd|JavaClassName> Optional. Command to be run as mapper.\n",
            "  -combiner       <cmd|JavaClassName> Optional. Command to be run as combiner.\n",
            "  -reducer        <cmd|JavaClassName> Optional. Command to be run as reducer.\n",
            "  -file           <file> Optional. File/dir to be shipped in the Job jar file.\n",
            "                  Deprecated. Use generic option \"-files\" instead.\n",
            "  -inputformat    <TextInputFormat(default)|SequenceFileAsTextInputFormat|JavaClassName>\n",
            "                  Optional. The input format class.\n",
            "  -outputformat   <TextOutputFormat(default)|JavaClassName>\n",
            "                  Optional. The output format class.\n",
            "  -partitioner    <JavaClassName>  Optional. The partitioner class.\n",
            "  -numReduceTasks <num> Optional. Number of reduce tasks.\n",
            "  -inputreader    <spec> Optional. Input recordreader spec.\n",
            "  -cmdenv         <n>=<v> Optional. Pass env.var to streaming commands.\n",
            "  -mapdebug       <cmd> Optional. To run this script when a map task fails.\n",
            "  -reducedebug    <cmd> Optional. To run this script when a reduce task fails.\n",
            "  -io             <identifier> Optional. Format to use for input to and output\n",
            "                  from mapper/reducer commands\n",
            "  -lazyOutput     Optional. Lazily create Output.\n",
            "  -background     Optional. Submit the job and don't wait till it completes.\n",
            "  -verbose        Optional. Print verbose output.\n",
            "  -info           Optional. Print detailed usage.\n",
            "  -help           Optional. Print help message.\n",
            "\n",
            "Generic options supported are:\n",
            "-conf <configuration file>        specify an application configuration file\n",
            "-D <property=value>               define a value for a given property\n",
            "-fs <file:///|hdfs://namenode:port> specify default filesystem URL to use, overrides 'fs.defaultFS' property from configurations.\n",
            "-jt <local|resourcemanager:port>  specify a ResourceManager\n",
            "-files <file1,...>                specify a comma-separated list of files to be copied to the map reduce cluster\n",
            "-libjars <jar1,...>               specify a comma-separated list of jar files to be included in the classpath\n",
            "-archives <archive1,...>          specify a comma-separated list of archives to be unarchived on the compute machines\n",
            "\n",
            "The general command line syntax is:\n",
            "command [genericOptions] [commandOptions]\n",
            "\n",
            "\n",
            "For more details about these options:\n",
            "Use $HADOOP_HOME/bin/hadoop jar hadoop-streaming.jar -info\n",
            "\n",
            "Try -help for more information\n",
            "Streaming Command Failed!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "%%bash\n",
        "hdfs dfs -rm -r my_output\n",
        "\n",
        "mapred streaming \\\n",
        "    -input hello.txt \\\n",
        "    -output my_output"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "5H2MkIUPyQc2",
        "outputId": "289335e1-bc89-4988-e57e-df5e6b15b563"
      },
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Deleted my_output\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "2024-03-24 19:44:30,756 INFO Configuration.deprecation: io.bytes.per.checksum is deprecated. Instead, use dfs.bytes-per-checksum\n",
            "2024-03-24 19:44:33,430 INFO impl.MetricsConfig: Loaded properties from hadoop-metrics2.properties\n",
            "2024-03-24 19:44:33,664 INFO impl.MetricsSystemImpl: Scheduled Metric snapshot period at 10 second(s).\n",
            "2024-03-24 19:44:33,664 INFO impl.MetricsSystemImpl: JobTracker metrics system started\n",
            "2024-03-24 19:44:33,690 WARN impl.MetricsSystemImpl: JobTracker metrics system already initialized!\n",
            "2024-03-24 19:44:34,018 INFO mapred.FileInputFormat: Total input files to process : 1\n",
            "2024-03-24 19:44:34,057 INFO mapreduce.JobSubmitter: number of splits:1\n",
            "2024-03-24 19:44:34,434 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_local225296935_0001\n",
            "2024-03-24 19:44:34,435 INFO mapreduce.JobSubmitter: Executing with tokens: []\n",
            "2024-03-24 19:44:34,757 INFO mapreduce.Job: The url to track the job: http://localhost:8080/\n",
            "2024-03-24 19:44:34,759 INFO mapreduce.Job: Running job: job_local225296935_0001\n",
            "2024-03-24 19:44:34,771 INFO mapred.LocalJobRunner: OutputCommitter set in config null\n",
            "2024-03-24 19:44:34,775 INFO mapred.LocalJobRunner: OutputCommitter is org.apache.hadoop.mapred.FileOutputCommitter\n",
            "2024-03-24 19:44:34,787 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:34,787 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:34,874 INFO mapred.LocalJobRunner: Waiting for map tasks\n",
            "2024-03-24 19:44:34,883 INFO mapred.LocalJobRunner: Starting task: attempt_local225296935_0001_m_000000_0\n",
            "2024-03-24 19:44:34,959 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:34,960 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:34,993 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]\n",
            "2024-03-24 19:44:35,008 INFO mapred.MapTask: Processing split: file:/content/hello.txt:0+14\n",
            "2024-03-24 19:44:35,032 INFO mapred.MapTask: numReduceTasks: 1\n",
            "2024-03-24 19:44:35,138 INFO mapred.MapTask: (EQUATOR) 0 kvi 26214396(104857584)\n",
            "2024-03-24 19:44:35,138 INFO mapred.MapTask: mapreduce.task.io.sort.mb: 100\n",
            "2024-03-24 19:44:35,138 INFO mapred.MapTask: soft limit at 83886080\n",
            "2024-03-24 19:44:35,138 INFO mapred.MapTask: bufstart = 0; bufvoid = 104857600\n",
            "2024-03-24 19:44:35,138 INFO mapred.MapTask: kvstart = 26214396; length = 6553600\n",
            "2024-03-24 19:44:35,145 INFO mapred.MapTask: Map output collector class = org.apache.hadoop.mapred.MapTask$MapOutputBuffer\n",
            "2024-03-24 19:44:35,157 INFO mapred.LocalJobRunner: \n",
            "2024-03-24 19:44:35,157 INFO mapred.MapTask: Starting flush of map output\n",
            "2024-03-24 19:44:35,157 INFO mapred.MapTask: Spilling map output\n",
            "2024-03-24 19:44:35,157 INFO mapred.MapTask: bufstart = 0; bufend = 22; bufvoid = 104857600\n",
            "2024-03-24 19:44:35,157 INFO mapred.MapTask: kvstart = 26214396(104857584); kvend = 26214396(104857584); length = 1/6553600\n",
            "2024-03-24 19:44:35,170 INFO mapred.MapTask: Finished spill 0\n",
            "2024-03-24 19:44:35,195 INFO mapred.Task: Task:attempt_local225296935_0001_m_000000_0 is done. And is in the process of committing\n",
            "2024-03-24 19:44:35,199 INFO mapred.LocalJobRunner: file:/content/hello.txt:0+14\n",
            "2024-03-24 19:44:35,199 INFO mapred.Task: Task 'attempt_local225296935_0001_m_000000_0' done.\n",
            "2024-03-24 19:44:35,210 INFO mapred.Task: Final Counters for attempt_local225296935_0001_m_000000_0: Counters: 17\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=141914\n",
            "\t\tFILE: Number of bytes written=852085\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tMap output bytes=22\n",
            "\t\tMap output materialized bytes=30\n",
            "\t\tInput split bytes=75\n",
            "\t\tCombine input records=0\n",
            "\t\tSpilled Records=1\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=0\n",
            "\t\tGC time elapsed (ms)=0\n",
            "\t\tTotal committed heap usage (bytes)=432013312\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "2024-03-24 19:44:35,211 INFO mapred.LocalJobRunner: Finishing task: attempt_local225296935_0001_m_000000_0\n",
            "2024-03-24 19:44:35,212 INFO mapred.LocalJobRunner: map task executor complete.\n",
            "2024-03-24 19:44:35,220 INFO mapred.LocalJobRunner: Waiting for reduce tasks\n",
            "2024-03-24 19:44:35,220 INFO mapred.LocalJobRunner: Starting task: attempt_local225296935_0001_r_000000_0\n",
            "2024-03-24 19:44:35,257 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:35,257 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:35,257 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]\n",
            "2024-03-24 19:44:35,267 INFO mapred.ReduceTask: Using ShuffleConsumerPlugin: org.apache.hadoop.mapreduce.task.reduce.Shuffle@582daec5\n",
            "2024-03-24 19:44:35,273 WARN impl.MetricsSystemImpl: JobTracker metrics system already initialized!\n",
            "2024-03-24 19:44:35,309 INFO reduce.MergeManagerImpl: MergerManager: memoryLimit=2382574336, maxSingleShuffleLimit=595643584, mergeThreshold=1572499072, ioSortFactor=10, memToMemMergeOutputsThreshold=10\n",
            "2024-03-24 19:44:35,325 INFO reduce.EventFetcher: attempt_local225296935_0001_r_000000_0 Thread started: EventFetcher for fetching Map Completion Events\n",
            "2024-03-24 19:44:35,386 INFO reduce.LocalFetcher: localfetcher#1 about to shuffle output of map attempt_local225296935_0001_m_000000_0 decomp: 26 len: 30 to MEMORY\n",
            "2024-03-24 19:44:35,393 INFO reduce.InMemoryMapOutput: Read 26 bytes from map-output for attempt_local225296935_0001_m_000000_0\n",
            "2024-03-24 19:44:35,398 INFO reduce.MergeManagerImpl: closeInMemoryFile -> map-output of size: 26, inMemoryMapOutputs.size() -> 1, commitMemory -> 0, usedMemory ->26\n",
            "2024-03-24 19:44:35,402 INFO reduce.EventFetcher: EventFetcher is interrupted.. Returning\n",
            "2024-03-24 19:44:35,404 INFO mapred.LocalJobRunner: 1 / 1 copied.\n",
            "2024-03-24 19:44:35,404 INFO reduce.MergeManagerImpl: finalMerge called with 1 in-memory map-outputs and 0 on-disk map-outputs\n",
            "2024-03-24 19:44:35,414 INFO mapred.Merger: Merging 1 sorted segments\n",
            "2024-03-24 19:44:35,415 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 16 bytes\n",
            "2024-03-24 19:44:35,417 INFO reduce.MergeManagerImpl: Merged 1 segments, 26 bytes to disk to satisfy reduce memory limit\n",
            "2024-03-24 19:44:35,418 INFO reduce.MergeManagerImpl: Merging 1 files, 30 bytes from disk\n",
            "2024-03-24 19:44:35,419 INFO reduce.MergeManagerImpl: Merging 0 segments, 0 bytes from memory into reduce\n",
            "2024-03-24 19:44:35,419 INFO mapred.Merger: Merging 1 sorted segments\n",
            "2024-03-24 19:44:35,420 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 16 bytes\n",
            "2024-03-24 19:44:35,421 INFO mapred.LocalJobRunner: 1 / 1 copied.\n",
            "2024-03-24 19:44:35,434 INFO mapred.Task: Task:attempt_local225296935_0001_r_000000_0 is done. And is in the process of committing\n",
            "2024-03-24 19:44:35,436 INFO mapred.LocalJobRunner: 1 / 1 copied.\n",
            "2024-03-24 19:44:35,436 INFO mapred.Task: Task attempt_local225296935_0001_r_000000_0 is allowed to commit now\n",
            "2024-03-24 19:44:35,439 INFO output.FileOutputCommitter: Saved output of task 'attempt_local225296935_0001_r_000000_0' to file:/content/my_output\n",
            "2024-03-24 19:44:35,440 INFO mapred.LocalJobRunner: reduce > reduce\n",
            "2024-03-24 19:44:35,441 INFO mapred.Task: Task 'attempt_local225296935_0001_r_000000_0' done.\n",
            "2024-03-24 19:44:35,441 INFO mapred.Task: Final Counters for attempt_local225296935_0001_r_000000_0: Counters: 24\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=142006\n",
            "\t\tFILE: Number of bytes written=852143\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tCombine input records=0\n",
            "\t\tCombine output records=0\n",
            "\t\tReduce input groups=1\n",
            "\t\tReduce shuffle bytes=30\n",
            "\t\tReduce input records=1\n",
            "\t\tReduce output records=1\n",
            "\t\tSpilled Records=1\n",
            "\t\tShuffled Maps =1\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=1\n",
            "\t\tGC time elapsed (ms)=0\n",
            "\t\tTotal committed heap usage (bytes)=432013312\n",
            "\tShuffle Errors\n",
            "\t\tBAD_ID=0\n",
            "\t\tCONNECTION=0\n",
            "\t\tIO_ERROR=0\n",
            "\t\tWRONG_LENGTH=0\n",
            "\t\tWRONG_MAP=0\n",
            "\t\tWRONG_REDUCE=0\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=28\n",
            "2024-03-24 19:44:35,442 INFO mapred.LocalJobRunner: Finishing task: attempt_local225296935_0001_r_000000_0\n",
            "2024-03-24 19:44:35,442 INFO mapred.LocalJobRunner: reduce task executor complete.\n",
            "2024-03-24 19:44:35,768 INFO mapreduce.Job: Job job_local225296935_0001 running in uber mode : false\n",
            "2024-03-24 19:44:35,769 INFO mapreduce.Job:  map 100% reduce 100%\n",
            "2024-03-24 19:44:35,771 INFO mapreduce.Job: Job job_local225296935_0001 completed successfully\n",
            "2024-03-24 19:44:35,783 INFO mapreduce.Job: Counters: 30\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=283920\n",
            "\t\tFILE: Number of bytes written=1704228\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tMap output bytes=22\n",
            "\t\tMap output materialized bytes=30\n",
            "\t\tInput split bytes=75\n",
            "\t\tCombine input records=0\n",
            "\t\tCombine output records=0\n",
            "\t\tReduce input groups=1\n",
            "\t\tReduce shuffle bytes=30\n",
            "\t\tReduce input records=1\n",
            "\t\tReduce output records=1\n",
            "\t\tSpilled Records=2\n",
            "\t\tShuffled Maps =1\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=1\n",
            "\t\tGC time elapsed (ms)=0\n",
            "\t\tTotal committed heap usage (bytes)=864026624\n",
            "\tShuffle Errors\n",
            "\t\tBAD_ID=0\n",
            "\t\tCONNECTION=0\n",
            "\t\tIO_ERROR=0\n",
            "\t\tWRONG_LENGTH=0\n",
            "\t\tWRONG_MAP=0\n",
            "\t\tWRONG_REDUCE=0\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=28\n",
            "2024-03-24 19:44:35,783 INFO streaming.StreamJob: Output directory: my_output\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Verify the result"
      ],
      "metadata": {
        "id": "v7Ks3e96yXuB"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%bash\n",
        "\n",
        "echo \"Check if MapReduce job was successful\"\n",
        "hdfs dfs -test -e my_output/_SUCCESS\n",
        "if [ $? -eq 0 ]; then\n",
        "\techo \"_SUCCESS exists!\"\n",
        "fi"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "cWAXvG0_yThc",
        "outputId": "4038a64c-db71-4a72-c6b6-fd2d9ba400e2"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Check if MapReduce job was successful\n",
            "_SUCCESS exists!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "Show output"
      ],
      "metadata": {
        "id": "t40GgJ2Hya9P"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!cat my_output/part-00000"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "I5APWEgoyaRS",
        "outputId": "119c1209-9837-4d47-d324-1493535482ce"
      },
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "0\tHello, World!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "What happened here is that not having defined any mapper or reducer, the \"Identity\" mapper ([IdentityMapper](https://hadoop.apache.org/docs/stable/api/org/apache/hadoop/mapred/lib/IdentityMapper.html)) and reducer ([IdentityReducer](https://hadoop.apache.org/docs/stable/api/org/apache/hadoop/mapred/lib/IdentityReducer.html)) were used by default (see [Streaming command options](https://hadoop.apache.org/docs/stable/hadoop-streaming/HadoopStreaming.html#Streaming_Command_Options))."
      ],
      "metadata": {
        "id": "mzfaMVKqyjpC"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Run a map-only MapReduce job\n",
        "\n",
        "Not specifying mapper and reducer in the MapReduce job submission does not mean that MapReduce isn't going to run the mapper and reducer steps, it is simply going to use the Identity mapper and reducer.\n",
        "\n",
        "To run a MapReduce job _without_ reducer one needs to use the generic option\n",
        "\n",
        "    \\-D mapreduce.job.reduces=0\n",
        "\n",
        "(see [specifying map-only jobs](https://hadoop.apache.org/docs/stable/hadoop-streaming/HadoopStreaming.html#Specifying_Map-Only_Jobs))."
      ],
      "metadata": {
        "id": "lzIuWv7Myndc"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%bash\n",
        "hdfs dfs -rm -r my_output\n",
        "\n",
        "mapred streaming \\\n",
        "    -D mapreduce.job.reduces=0 \\\n",
        "    -input hello.txt \\\n",
        "    -output my_output"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "OdwKWyVRye27",
        "outputId": "1a588516-c218-431a-bf26-72cd75863791"
      },
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Deleted my_output\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "2024-03-24 19:44:40,243 INFO Configuration.deprecation: io.bytes.per.checksum is deprecated. Instead, use dfs.bytes-per-checksum\n",
            "2024-03-24 19:44:44,191 INFO impl.MetricsConfig: Loaded properties from hadoop-metrics2.properties\n",
            "2024-03-24 19:44:44,410 INFO impl.MetricsSystemImpl: Scheduled Metric snapshot period at 10 second(s).\n",
            "2024-03-24 19:44:44,411 INFO impl.MetricsSystemImpl: JobTracker metrics system started\n",
            "2024-03-24 19:44:44,441 WARN impl.MetricsSystemImpl: JobTracker metrics system already initialized!\n",
            "2024-03-24 19:44:44,755 INFO mapred.FileInputFormat: Total input files to process : 1\n",
            "2024-03-24 19:44:44,842 INFO mapreduce.JobSubmitter: number of splits:1\n",
            "2024-03-24 19:44:45,207 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_local1085280640_0001\n",
            "2024-03-24 19:44:45,207 INFO mapreduce.JobSubmitter: Executing with tokens: []\n",
            "2024-03-24 19:44:45,472 INFO mapreduce.Job: The url to track the job: http://localhost:8080/\n",
            "2024-03-24 19:44:45,475 INFO mapreduce.Job: Running job: job_local1085280640_0001\n",
            "2024-03-24 19:44:45,486 INFO mapred.LocalJobRunner: OutputCommitter set in config null\n",
            "2024-03-24 19:44:45,490 INFO mapred.LocalJobRunner: OutputCommitter is org.apache.hadoop.mapred.FileOutputCommitter\n",
            "2024-03-24 19:44:45,503 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:45,503 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:45,611 INFO mapred.LocalJobRunner: Waiting for map tasks\n",
            "2024-03-24 19:44:45,622 INFO mapred.LocalJobRunner: Starting task: attempt_local1085280640_0001_m_000000_0\n",
            "2024-03-24 19:44:45,676 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:45,679 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:45,731 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]\n",
            "2024-03-24 19:44:45,745 INFO mapred.MapTask: Processing split: file:/content/hello.txt:0+14\n",
            "2024-03-24 19:44:45,764 INFO mapred.MapTask: numReduceTasks: 0\n",
            "2024-03-24 19:44:45,793 INFO mapred.LocalJobRunner: \n",
            "2024-03-24 19:44:45,807 INFO mapred.Task: Task:attempt_local1085280640_0001_m_000000_0 is done. And is in the process of committing\n",
            "2024-03-24 19:44:45,810 INFO mapred.LocalJobRunner: \n",
            "2024-03-24 19:44:45,810 INFO mapred.Task: Task attempt_local1085280640_0001_m_000000_0 is allowed to commit now\n",
            "2024-03-24 19:44:45,813 INFO output.FileOutputCommitter: Saved output of task 'attempt_local1085280640_0001_m_000000_0' to file:/content/my_output\n",
            "2024-03-24 19:44:45,818 INFO mapred.LocalJobRunner: file:/content/hello.txt:0+14\n",
            "2024-03-24 19:44:45,818 INFO mapred.Task: Task 'attempt_local1085280640_0001_m_000000_0' done.\n",
            "2024-03-24 19:44:45,829 INFO mapred.Task: Final Counters for attempt_local1085280640_0001_m_000000_0: Counters: 15\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=141914\n",
            "\t\tFILE: Number of bytes written=855489\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tInput split bytes=75\n",
            "\t\tSpilled Records=0\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=0\n",
            "\t\tGC time elapsed (ms)=19\n",
            "\t\tTotal committed heap usage (bytes)=341835776\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=28\n",
            "2024-03-24 19:44:45,830 INFO mapred.LocalJobRunner: Finishing task: attempt_local1085280640_0001_m_000000_0\n",
            "2024-03-24 19:44:45,831 INFO mapred.LocalJobRunner: map task executor complete.\n",
            "2024-03-24 19:44:46,483 INFO mapreduce.Job: Job job_local1085280640_0001 running in uber mode : false\n",
            "2024-03-24 19:44:46,487 INFO mapreduce.Job:  map 100% reduce 0%\n",
            "2024-03-24 19:44:46,494 INFO mapreduce.Job: Job job_local1085280640_0001 completed successfully\n",
            "2024-03-24 19:44:46,503 INFO mapreduce.Job: Counters: 15\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=141914\n",
            "\t\tFILE: Number of bytes written=855489\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tInput split bytes=75\n",
            "\t\tSpilled Records=0\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=0\n",
            "\t\tGC time elapsed (ms)=19\n",
            "\t\tTotal committed heap usage (bytes)=341835776\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=28\n",
            "2024-03-24 19:44:46,503 INFO streaming.StreamJob: Output directory: my_output\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Verify the result"
      ],
      "metadata": {
        "id": "QZIE9yXOyyHJ"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!hdfs dfs -test -e my_output/_SUCCESS && cat my_output/part-00000"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "7Dt3tUI0yu5e",
        "outputId": "5d7ac539-448c-48cf-994e-1e27ac88f979"
      },
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "0\tHello, World!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Why a map-only application?\n",
        "\n",
        "The advantage of a map-only job is that the sorting and shuffling phases are skipped, so if you do not need that remember to specify `-D mapreduce.job.reduces=0 `.\n",
        "\n",
        "On the other hand, a MapReduce job even with the default `IdentityReducer` will deliver sorted results because the data passed from the mapper to the reducer always gets sorted.\n"
      ],
      "metadata": {
        "id": "hUGEUv99y3cM"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Improved version of the MapReduce \"Hello, World!\" application\n",
        "\n",
        "Taking into account the previous considerations, here's a more efficient version of the 'Hello, World!' application that bypasses the shuffling and sorting step."
      ],
      "metadata": {
        "id": "FhVVFEdKzGcI"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%bash\n",
        "hdfs dfs -rm -r my_output\n",
        "\n",
        "mapred streaming \\\n",
        "    -D mapreduce.job.reduces=0 \\\n",
        "    -input hello.txt \\\n",
        "    -output my_output \\\n",
        "    -mapper '/bin/cat'"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "jLgMXX2jy0vC",
        "outputId": "0a4c9e2c-8575-4097-b360-8cd549091244"
      },
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Deleted my_output\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "2024-03-24 19:44:50,920 INFO Configuration.deprecation: io.bytes.per.checksum is deprecated. Instead, use dfs.bytes-per-checksum\n",
            "2024-03-24 19:44:53,832 INFO impl.MetricsConfig: Loaded properties from hadoop-metrics2.properties\n",
            "2024-03-24 19:44:54,249 INFO impl.MetricsSystemImpl: Scheduled Metric snapshot period at 10 second(s).\n",
            "2024-03-24 19:44:54,249 INFO impl.MetricsSystemImpl: JobTracker metrics system started\n",
            "2024-03-24 19:44:54,310 WARN impl.MetricsSystemImpl: JobTracker metrics system already initialized!\n",
            "2024-03-24 19:44:54,927 INFO mapred.FileInputFormat: Total input files to process : 1\n",
            "2024-03-24 19:44:55,003 INFO mapreduce.JobSubmitter: number of splits:1\n",
            "2024-03-24 19:44:55,602 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_local840123681_0001\n",
            "2024-03-24 19:44:55,602 INFO mapreduce.JobSubmitter: Executing with tokens: []\n",
            "2024-03-24 19:44:56,072 INFO mapreduce.Job: The url to track the job: http://localhost:8080/\n",
            "2024-03-24 19:44:56,084 INFO mapred.LocalJobRunner: OutputCommitter set in config null\n",
            "2024-03-24 19:44:56,096 INFO mapreduce.Job: Running job: job_local840123681_0001\n",
            "2024-03-24 19:44:56,104 INFO mapred.LocalJobRunner: OutputCommitter is org.apache.hadoop.mapred.FileOutputCommitter\n",
            "2024-03-24 19:44:56,124 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:56,125 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:56,271 INFO mapred.LocalJobRunner: Waiting for map tasks\n",
            "2024-03-24 19:44:56,294 INFO mapred.LocalJobRunner: Starting task: attempt_local840123681_0001_m_000000_0\n",
            "2024-03-24 19:44:56,370 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 2\n",
            "2024-03-24 19:44:56,373 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
            "2024-03-24 19:44:56,404 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]\n",
            "2024-03-24 19:44:56,417 INFO mapred.MapTask: Processing split: file:/content/hello.txt:0+14\n",
            "2024-03-24 19:44:56,436 INFO mapred.MapTask: numReduceTasks: 0\n",
            "2024-03-24 19:44:56,455 INFO streaming.PipeMapRed: PipeMapRed exec [/bin/cat]\n",
            "2024-03-24 19:44:56,491 INFO Configuration.deprecation: mapred.work.output.dir is deprecated. Instead, use mapreduce.task.output.dir\n",
            "2024-03-24 19:44:56,501 INFO Configuration.deprecation: mapred.local.dir is deprecated. Instead, use mapreduce.cluster.local.dir\n",
            "2024-03-24 19:44:56,501 INFO Configuration.deprecation: map.input.file is deprecated. Instead, use mapreduce.map.input.file\n",
            "2024-03-24 19:44:56,502 INFO Configuration.deprecation: map.input.length is deprecated. Instead, use mapreduce.map.input.length\n",
            "2024-03-24 19:44:56,503 INFO Configuration.deprecation: mapred.job.id is deprecated. Instead, use mapreduce.job.id\n",
            "2024-03-24 19:44:56,503 INFO Configuration.deprecation: mapred.task.partition is deprecated. Instead, use mapreduce.task.partition\n",
            "2024-03-24 19:44:56,505 INFO Configuration.deprecation: map.input.start is deprecated. Instead, use mapreduce.map.input.start\n",
            "2024-03-24 19:44:56,505 INFO Configuration.deprecation: mapred.task.is.map is deprecated. Instead, use mapreduce.task.ismap\n",
            "2024-03-24 19:44:56,506 INFO Configuration.deprecation: mapred.task.id is deprecated. Instead, use mapreduce.task.attempt.id\n",
            "2024-03-24 19:44:56,507 INFO Configuration.deprecation: mapred.tip.id is deprecated. Instead, use mapreduce.task.id\n",
            "2024-03-24 19:44:56,516 INFO Configuration.deprecation: mapred.skip.on is deprecated. Instead, use mapreduce.job.skiprecords\n",
            "2024-03-24 19:44:56,517 INFO Configuration.deprecation: user.name is deprecated. Instead, use mapreduce.job.user.name\n",
            "2024-03-24 19:44:56,546 INFO streaming.PipeMapRed: R/W/S=1/0/0 in:NA [rec/s] out:NA [rec/s]\n",
            "2024-03-24 19:44:56,549 INFO streaming.PipeMapRed: MRErrorThread done\n",
            "2024-03-24 19:44:56,550 INFO streaming.PipeMapRed: Records R/W=1/1\n",
            "2024-03-24 19:44:56,550 INFO streaming.PipeMapRed: mapRedFinished\n",
            "2024-03-24 19:44:56,554 INFO mapred.LocalJobRunner: \n",
            "2024-03-24 19:44:56,566 INFO mapred.Task: Task:attempt_local840123681_0001_m_000000_0 is done. And is in the process of committing\n",
            "2024-03-24 19:44:56,568 INFO mapred.LocalJobRunner: \n",
            "2024-03-24 19:44:56,568 INFO mapred.Task: Task attempt_local840123681_0001_m_000000_0 is allowed to commit now\n",
            "2024-03-24 19:44:56,573 INFO output.FileOutputCommitter: Saved output of task 'attempt_local840123681_0001_m_000000_0' to file:/content/my_output\n",
            "2024-03-24 19:44:56,575 INFO mapred.LocalJobRunner: Records R/W=1/1\n",
            "2024-03-24 19:44:56,575 INFO mapred.Task: Task 'attempt_local840123681_0001_m_000000_0' done.\n",
            "2024-03-24 19:44:56,587 INFO mapred.Task: Final Counters for attempt_local840123681_0001_m_000000_0: Counters: 15\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=141914\n",
            "\t\tFILE: Number of bytes written=855001\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tInput split bytes=75\n",
            "\t\tSpilled Records=0\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=0\n",
            "\t\tGC time elapsed (ms)=20\n",
            "\t\tTotal committed heap usage (bytes)=371195904\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=27\n",
            "2024-03-24 19:44:56,587 INFO mapred.LocalJobRunner: Finishing task: attempt_local840123681_0001_m_000000_0\n",
            "2024-03-24 19:44:56,589 INFO mapred.LocalJobRunner: map task executor complete.\n",
            "2024-03-24 19:44:57,115 INFO mapreduce.Job: Job job_local840123681_0001 running in uber mode : false\n",
            "2024-03-24 19:44:57,117 INFO mapreduce.Job:  map 100% reduce 0%\n",
            "2024-03-24 19:44:57,121 INFO mapreduce.Job: Job job_local840123681_0001 completed successfully\n",
            "2024-03-24 19:44:57,130 INFO mapreduce.Job: Counters: 15\n",
            "\tFile System Counters\n",
            "\t\tFILE: Number of bytes read=141914\n",
            "\t\tFILE: Number of bytes written=855001\n",
            "\t\tFILE: Number of read operations=0\n",
            "\t\tFILE: Number of large read operations=0\n",
            "\t\tFILE: Number of write operations=0\n",
            "\tMap-Reduce Framework\n",
            "\t\tMap input records=1\n",
            "\t\tMap output records=1\n",
            "\t\tInput split bytes=75\n",
            "\t\tSpilled Records=0\n",
            "\t\tFailed Shuffles=0\n",
            "\t\tMerged Map outputs=0\n",
            "\t\tGC time elapsed (ms)=20\n",
            "\t\tTotal committed heap usage (bytes)=371195904\n",
            "\tFile Input Format Counters \n",
            "\t\tBytes Read=14\n",
            "\tFile Output Format Counters \n",
            "\t\tBytes Written=27\n",
            "2024-03-24 19:44:57,130 INFO streaming.StreamJob: Output directory: my_output\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!hdfs dfs -test -e my_output/_SUCCESS && cat my_output/part-00000"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Sa1UDPr6zKKw",
        "outputId": "8646796f-1332-4515-9188-2b5f217ada34"
      },
      "execution_count": 17,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Hello, World!\t\n"
          ]
        }
      ]
    }
  ]
}