015_hadoop.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "<a href=\"https://colab.research.google.com/github/rzl-ds/gu511/blob/master/015_hadoop.ipynb\" target=\"_parent\">\n",
    "    <img src=\"https://colab.research.google.com/assets/colab-badge.svg\"/>\n",
    "</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# `hadoop`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## `emr` spinup\n",
    "\n",
    "we will want to work with an `emr` `hadoop` cluster throughout this lecture, but it takes about 12 or so minutes to fully spin up. Normally, I would say: Just start it!\n",
    "\n",
    "But as this class is async, quick note of caution.\n",
    "\n",
    "**this cluster will cost 0.745 USD**\n",
    "\n",
    "if you just want it on during your watching of the lecture, go for it. However, if you want to pause, watch part today and part tomorrow, etc -- you may want to wait to start your cluster until you need it.\n",
    "\n",
    "and on the other end: **be sure you shut off your cluster when you are done, or if you aren't going to use it for several hours**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">starting up an `emr` cluster</div>**\n",
    "\n",
    "the below steps will create a cluster which will cost about 0.74 USD per hour.\n",
    "\n",
    "+ in the `aws` web console open the `emr` service\n",
    "+ click create cluster, and on the \"quick options\" screen select \"advanced options\"\n",
    "+ software and steps\n",
    "    + stay at emr-5.31.0\n",
    "    + for software, click:\n",
    "        + `hadoop`\n",
    "        + `ganglia`\n",
    "        + `hive`\n",
    "        + `hue`\n",
    "        + `spark`\n",
    "        + `livy`\n",
    "        + `pig`\n",
    "    + notice but don't click: `jupyterhub`, `mxnet`, `tensorflow`\n",
    "    + click next\n",
    "+ hardware config -- leave all defaults\n",
    "    + generally, think about space requirements [ala this](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-instances-guidelines.html#emr-plan-instances-hdfs)\n",
    "+ general options -- pick a name\n",
    "+ security options -- ***choose a key pair! make sure you have that key!!***"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## the problem(s)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "reach all the way back in your memory several lectures ago, when we talked about `nosql` databases, and in particular document stores like `aws` `dynamodb`. the proposed use case for these `nosql` databases was an ambiguous \"webby\" one:\n",
    "\n",
    "*we're reading and writing way too much data way too fast for our one machine*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "this sentiment is reflective of a modern data reality that often goes by the buzziest of buzzwords:\n",
    "\n",
    "**big data**\n",
    "\n",
    "*mandatory caveat: big data $\\neq$ data science*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "traditional data analyses were optimized for super-powerful single machines such as the monolithic, super-powerful `sql` servers\n",
    "\n",
    "*note: this is not to say that cluster computing didn't exist; indeed it was one of the main computational frameworks in the early days of computers*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "our exponential growth in disk space and memory space per dollar spent fueled a lot of this work and innovation.\n",
    "\n",
    "basically, for a long while our ability to *compute* data grew faster than our ability to *create* or *acquire* data. in the modern world, though, that notion is absolute history."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "take, for example, a relatively trivial process for modern computation: word counts for a document of several MBs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "rm /tmp/shakespeare.txt 2> /dev/null\n",
    "wget --quiet -O /tmp/shakespeare.txt.zip https://github.com/bbengfort/hadoop-fundamentals/raw/master/data/shakespeare.txt.zip\n",
    "unzip /tmp/shakespeare.txt.zip -d /tmp\n",
    "ls -alh /tmp/shak*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "head /tmp/shakespeare.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "with open('/tmp/shakespeare.txt', 'r') as f:\n",
    "    s = f.read()\n",
    "\n",
    "print(s[:100])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import collections\n",
    "\n",
    "wordct = collections.Counter(\n",
    "    word.lower()\n",
    "    for line in s.split('\\n')\n",
    "    for word in line.strip().split('\\t')\n",
    "    if word)\n",
    "\n",
    "wordct.most_common(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "*don't mind me, just cleaning up...*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!rm /tmp/shakespeare.txt*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "that was easy, but it relied on some important features:\n",
    "\n",
    "1. I had enough disk space to have that 8.5MB file stored locally\n",
    "2. I had enough memory to load that 8.5MB file's contents directly into memory"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "obviously, that isn't always the case. It's not even hard to think of counter-examples\n",
    "\n",
    "1. a larger text corpus (e.g. all of wikipedia, 10TB as of 2015, or publicly available SEC EDGAR filings)\n",
    "2. any reasonably large image recognition project\n",
    "3. the logs of web traffic for any modestly sized website or service\n",
    "4. IoT information (usage records of your smartphone or headphones, e.g.)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "so, back to `dynamodb`. when we (theoretically) started to run into resource issues for our single-machine architecture, we decided to change the way we were doing things and start \"scaling horizontally\" -- choose an architecture and software that can spread the storage and computation burden across multiple machines"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "for `dynamodb` we were attempting to distribute out our database writes and reads as actions, but the test scenario I laid out above was one of\n",
    "\n",
    "+ data storage\n",
    "+ resource availability"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`hadoop` is a distributed storage and computing software, and although it's dominance is waning (especially as `spark` takes over many of its use cases), it is still the primary big data distributed storage and ETL processing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "it is a software solution that abstracts out all the \"hard stuff\" (the complicated networking and resource marshaling) that needs to happen to get multiple computers on the same page, and instead provides the user (you) with a single api for\n",
    "\n",
    "+ accessing distributed files (`hdfs`)\n",
    "+ securing computational resources and memory (`yarn`)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## `hadoop` nuts and bolts"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "let's dig into the details of distributed computing a bit"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### terminology\n",
    "\n",
    "+ **node**: a single machine (real or virtual)\n",
    "+ **cluster**: a collection of *nodes* which can communicate with each other\n",
    "+ **master**: a *node* which can request information from or delegate tasks to other *nodes*\n",
    "+ **worker**: a *node* which merely receives, processes, and responds to requests"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "+ **local**\n",
    "\n",
    "up until now, we have cultivated a healthy habit of calling our laptop \"local\" and our `ec2` instance \"remote\", but we will break that temporarily here to talk about `hdfs`.\n",
    "\n",
    "in the `hadoop` context, when we say \"local\" we generally mean whatever machine we are on when we are executing `hadoop` commands. note that this can be our (up to now exclusively \"remote\") `ec2` server."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### basic concepts\n",
    "\n",
    "in the database world we had certain requirements a `dbms` needed to meet to ensure that all clients of that database service could share those resources (the `ACID` principles)\n",
    "\n",
    "similarly, for distributed computing to be well defined and robust, we have four requirements:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "1. *fault tolerance*: if one computer goes down, we're good. if it comes back, we're even gooder\n",
    "2. *recoverability*: we don't lose data when things fail\n",
    "3. *consistency*: results shouldn't depend on jobs failing or succeeding\n",
    "4. *scalability*: more data means longer time, not failure; we can increase resources if desired"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`hadoop` addresses these requirements by making the following decisions:\n",
    "\n",
    "+ data is distributed across many nodes in the cluster; each node prefers its local data\n",
    "+ all data is chunked into blocks (say, 128 MB) and is *replicated* (copied to other nodes)\n",
    "\n",
    "<img src=\"https://d1jnx9ba8s6j9r.cloudfront.net/blog/wp-content/uploads/2016/10/Replication-Management-Apache-Hadoop-HDFS-Architecture-Edureka-Blog.png\" width=\"800px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "+ jobs (computations) are broken into tasks applied to single blocks\n",
    "+ jobs are completely unaware that they are distributed\n",
    "+ worker nodes don't care about each other\n",
    "+ tasks are redundant and repeatable\n",
    "+ master nodes handle allocation of all resources (storage, cpus, memory)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `hadoop` architecture\n",
    "\n",
    "`hadoop` is open source software which defines a number of utilities, but the two most important are:\n",
    "\n",
    "1. `hdfs` (a program for handling distributed file storage)\n",
    "2. `yarn` (a program for handling distributed resource allocation)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "together, these two services work together to enforce some of those design decisions above: namely, to make sure that all data is robustly distributed and that all distributed tasks are working on local data.\n",
    "\n",
    "`hdfs` handles the distribution of files (e.g. how is data chunked into blocks? where did I leave block 1337? who do I ask to send that block to me?)\n",
    "\n",
    "`yarn` handles the distribution of tasks, aka requests for compute resources (e.g. I was asked to calculate a word count, how do I break this task up in a way my lovable but dumb worker nodes will understand?)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`hdfs` and `yarn` are the default software services for managing distributed storage or compute resources - and they were specifically designed to work in tandem - but either can be replaced or augmented:\n",
    "\n",
    "+ you could change storage methods (e.g. `hdfs` replaced by `s3` or `dbfs`)\n",
    "+ you could change resource managers or computational layers on top of storage (e.g. `yarn` replaced by `spark` or `mesos`)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### a `hadoop` cluster\n",
    "\n",
    "`hadoop` is a software. the hardware is a cluster of computers. the benefit you get in using `hdfs` and `yarn` are abstracted `api`s that hide cluster administration details and tasks from you.\n",
    "\n",
    "to put it another way: `hadoop` lets someone else do the hard task of distribution so you can do what you came here to do (data processing).\n",
    "\n",
    "the cost of deferring that hard work is that you must fit your analysis into that abstracted `api` -- not always easy!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "in this class, we've seen many instances of the client / service paradigm (e.g. when making `ssh` connections, when making `REST` requests, when connecting to databases)\n",
    "\n",
    "both `hdfs` and `yarn` have several *services* that our tools (*clients*) use"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`hdfs` services:\n",
    "\n",
    "+ `NameNode` (master): stores the directory tree, file metadata, file cluster locations. this is the access point for `hdfs` usage\n",
    "+ `Secondary NameNode` (master): performs housekeeping, checkpointing. this is more like an assistant `NameNode` than a backup `NameNode`\n",
    "+ `DataNode` (worker): local `io` for `hdfs` blocks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the basic interaction with `hdfs`:\n",
    "\n",
    "1. client asks `NameNode` where data lives.\n",
    "2. `NameNode` tells client\n",
    "3. client is responsible for going and getting data from `DataNode`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`yarn` services:\n",
    "\n",
    "+ `ResourceManager` (master): allocates and monitor resources (memory, cores), schedules jobs\n",
    "+ `ApplicationMaster` (master): coordinates a particular app after `ResourceManager` has scheduled it\n",
    "+ `NodeManager` (worker): runs tasks and reports on task status"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the basic interaction with `yarn` is very similar:\n",
    "\n",
    "1. client asks `ResourceManager` for resources\n",
    "2. `ResourceManager` assigns `ApplicationMaster` instance to manage the individual application\n",
    "3. `ApplicationMaster` submits a job to a single `NodeManager`, tracks all submitted jobs\n",
    "4. `NodeManager` executes incoming assigned tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "to give you a sense of scale for a typical `hadoop` cluster\n",
    "\n",
    "+ 20 - 30 workers and one master can handle 10s of terabytes of data in simultaneous workflows\n",
    "+ to get into hundreds of nodes at a time, you will need to start migrating master services to separate servers (no more single master nodes)\n",
    "+ to get into the thousands of nodes, you will need to create multiple master service machines just for communication between the nodes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### details on `hdfs`\n",
    "\n",
    "`hdfs` is a file system on top of another filesystem. in many respects, it behaves like you're used to the `linux` filesystem behaving (with slightly different commands). there are a few nuances worth discussing, however."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "##### blocks\n",
    "\n",
    "files are blocked into large (e.g. 128MB) chunks. this means that a file larger than that will be chopped up into different blocks."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "for example, if I have a 450 MB file `corpus.txt` and a block size of 128 MB per block, `hadoop` will break up that `corpus.txt` file into chunks and distribute them (perhaps with replication) to different workers. schematically (the names don't exist in real life):\n",
    "\n",
    "+ `corpus.txt.1` (0 - 128 MB)\n",
    "+ `corpus.txt.2` (128 - 256 MB)\n",
    "+ `corpus.txt.3` (256 - 384 MB)\n",
    "+ `corpus.txt.4` (384 - 450 MB)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "it's worth noting: unlike block sizes in non-distributed file systems, in `hadoop` a small file will not wastefully take up the remainder of the space on the hard drive."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "that's not to say there isn't a problem with small files, though -- there is. It's not wasteful disk usage, it's wasteful *resource* usage. we will discuss `mappers` and `reducers` later, but for now it suffices to say: when we distributed tasks, we already said we distribute them to each needed block.\n",
    "\n",
    "if we have many small files we will have to keep track of many different blocks, and every time we perform a task we'll have to create many sub-tasks to do so. that will be pretty wasteful."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "in the end: better a million files of 100 MB than a billion files of 0.1 MB"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">PAUSE FOR ZOOM BREAK</div>**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## `aws` `emr`: amazon's `hadoop`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "so you have decided you have enough data you need to move to a distributed environment -- you could buy several servers and install `hadoop` on them (a great exercise for masochists!)... or you could pay `aws` to do that for you. that's obviously what we've done.\n",
    "\n",
    "we stressed earlier that `hadoop` is a software, not a hardware.\n",
    "\n",
    "one implementation of that software (really, a modification) is the `aws` version of `hadoop`, which is called `e`lastic `m`ap `r`educe, or `emr`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "at the start of the class we created a `hadoop` cluster using the `aws` `emr` service -- it should be done by now, so let's go back and re-visit what we did."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "we asked amazon to create a 3-node cluster with\n",
    "\n",
    "+ 1 master node (running the `NameNode` `hdfs` services and the `Resource-` and `ApplicationManager` `yarn` services)\n",
    "+ 2 **core** nodes (running the `DataNode` `hdfs` service and the `NodeManager` `yarn` service)\n",
    "    + in `aws`, *core* nodes have `hdfs` blocks. all tasks using `hdfs` will rely on the *core* nodes\n",
    "+ 0 **task** nodes\n",
    "    + in `aws`, *task* nodes **do not** have `hdfs` blocks on them, and are instead available largely for computation purposes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "additionally, being an `aws` product, there is immediate integration with other `aws` services, and in particular `s3`.\n",
    "\n",
    "`emr` can use `hdfs` (in `aws` land, that is ephemeral storage that is destroyed when a cluster is taken down), or it can use `s3`, or it can use the local file systems on the `aws` `ec2` servers which are actually running the cluster.\n",
    "\n",
    "generally, input and output is done in `s3` and intermediate steps are retained in the ephemeral `hdfs` storage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "we also asked amazon to install several common `hadoop` technologies. we will explain them as we use them throughout the lecture"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### pricing\n",
    "\n",
    "`emr` pricing is not immediately obvious, but is simple enough. you pay a certain fee for the `emr` service depending on the underlying machine type you choose (e.g. `m5.xlarge`), and you *also* pay for those instances themselves (as if they were just `ec2` nodes you started), and you *also* pay for the attached `ebs` disk space (it's not a bundled deal!)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "for our demo we put together above, we made a 3-node cluster with `m5.xlarge` instances. as of writing, the rates for this `instance` type are:\n",
    "\n",
    "+ compute costs\n",
    "    + 0.192 USD per hour for each instance\n",
    "    + 0.048 USD per hour for the `emr` service for each instance\n",
    "    + three instances means `3 * (0.192 + 0.048) = 0.72`, 0.72 USD per hour\n",
    "+ storage costs\n",
    "    + 0.10 USD per GB-month\n",
    "        + given that we have 192 GB total (64 GB EBS for each machine by default), this is about 0.0258 USD per hour\n",
    "\n",
    "all total we're seeing 0.745 USD per hour. 3 hours is around 2.24 USD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "we're burning money!\n",
    "\n",
    "let's use our clusters already. log in to the `master` node as you would any other `ec2` instance: via `ssh`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">logging in to your `emr` cluster</div>**\n",
    "\n",
    "+ in the `aws` web console open the `emr` service and select your `emr` cluster\n",
    "+ update the security group of your master node to have your (or all) public ip addresses\n",
    "    + click on the security group linked under \"security groups for master\"\n",
    "    + there should be two records on this page -- click on the `master` group\n",
    "    + add an *inbound* rule allowing `ssh` type traffic from all ip address\n",
    "+ on the \"summary\" tab, find the value of \"Master public DNS\"\n",
    "    + e.g. mine was `ec2-34-231-180-183.compute-1.amazonaws.com`\n",
    "+ using the key pair you specified when you created the cluster, execute (replacing the `####` characters with your dns value we just grabbed)\n",
    "    + `ssh -i /path/to/your/keypair.pem hadoop@ec2-###-###-###-###.compute-1.amazonaws.com`\n",
    "    + note user is `hadoop` here, not `ubuntu` or `ec2-user`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## working with a distributed file system"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### basic file system operations\n",
    "\n",
    "many of the common `linux` command line file system tools are available with the same names in `hadoop`. try\n",
    "\n",
    "```bash\n",
    "hadoop fs -help\n",
    "```\n",
    "\n",
    "(note the single-dash parameters and curse the `java` gods)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "tired of reading those 4000 lines? try any one subcommand too:\n",
    "\n",
    "```bash\n",
    "hadoop fs -help ls\n",
    "hadoop fs -help chmod\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "let's prepare our `hadoop` cluster to actually do some `hadoop`-y stuff. on the `emr` master node, run:\n",
    "\n",
    "```bash\n",
    "# get the corpus locally\n",
    "mkdir ~/code && cd ~/code\n",
    "sudo yum install -y git\n",
    "git clone https://github.com/bbengfort/hadoop-fundamentals.git\n",
    "cd hadoop-fundamentals/data\n",
    "unzip shakespeare.txt.zip\n",
    "\n",
    "# put it into hdfs\n",
    "hadoop fs -mkdir /data\n",
    "hadoop fs -put shakespeare.txt /data/shakespeare.txt\n",
    "hadoop fs -ls -h /data\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the opposite of `put`-ting a file is `get`-ting a file, in which we can `get` items from `hdfs` and download them to our local hard drive\n",
    "\n",
    "```sh\n",
    "hadoop fs -get /data/shakespeare.txt /tmp/test_get.txt\n",
    "ls -alh /tmp/test*\n",
    "rm /tmp/test_get.txt\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the files we save in `hadoop` are generally enormous. it's good to know right away how to read portions of such large files:\n",
    "\n",
    "+ `hadoop fs -cat /data/shakespeare.txt | less`\n",
    "+ `hadoop fs -cat /data/shakespeare.txt | head` (this aborts the streaming when `head` has had enough)\n",
    "+ `hadoop fs -tail /data/shakespeare.txt`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### quick refresher on `uri`s and `schema`s\n",
    "\n",
    "above we executed our first `hdfs` file system operation -- a `put`. the `put` command will take a source file (assumed to be *local*, on your regular `emr` master node's hard drive) and will `put` it into a destination path in `hdfs`.\n",
    "\n",
    "we are talking about a path on the *local* file system and then a path in `hdfs` and using the same syntax -- just `/`-separated paths. we are relying on the `put` command *knowing* which is which"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "generally speaking, we may need to be more explicit than this. recall from the database class that we can uniquely describe any resource (`REST` api endpoints, files, etc) with a `url` or `uri` that includes a schema, a la\n",
    "\n",
    "```\n",
    "schema://[user[:password]@]host[:port]/path\n",
    "```\n",
    "\n",
    "secretly, `hadoop` has been converting all the arguments we've been giving it into `uri`s"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "for both the local (`ec2`) file system and `hdfs`, the `user`, `password`, and `port` don't have to be provided (this is always the case for the local `ec2` file system, and handled for `hdfs` by the way that `hadoop` has been configured for us by `aws`). furthermore, `hadoop` uses the `HOSTNAME` environment variable for the host value if none is provided. this collapses our `uri`s to\n",
    "\n",
    "```\n",
    "schema:///path\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "for files we want to access, the schemas are\n",
    "\n",
    "+ local file system: `file`\n",
    "+ `hdfs`: `hdfs`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "therefore, to describe the local file, the full path would be\n",
    "\n",
    "```\n",
    "file:///home/hadoop/code/hadoop-fundamentals/data/shakespeare.txt\n",
    "```\n",
    "\n",
    "and the `hdfs` full path would be\n",
    "\n",
    "```\n",
    "hdfs:///data/shakespeare.txt\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "to verify that the `HOSTNAME` value is used by the `hadoop` command, try:\n",
    "\n",
    "```sh\n",
    "hadoop fs -tail hdfs://$HOSTNAME/data/shakespeare.txt\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### a tangent: `hdfs dfs` vs. `hadoop fs`\n",
    "\n",
    "floating around out on the etherwebs, you may see stack overflow posts with commands such as\n",
    "\n",
    "```bash\n",
    "hdfs dfs -ls /\n",
    "```\n",
    "\n",
    "note that we're not writing that here, but instead writing `hadoop fs` instead of `hdfs dfs`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`hdfs dfs` is *related to* `hadoop fs`, but is not exactly the same. `hadoop fs` defaults to looking at `hdfs` files, but is actually file-system agnostic(ish), and supports local files (via the `file://` schema), `s3` files, `ftp` services, and any other schema people have been kind enough to implement.\n",
    "\n",
    "`hdfs dfs` *only* works with `hdfs`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "to demonstrate how `hadoop fs` can be used with local files as well, try out\n",
    "\n",
    "```bash\n",
    "hadoop fs -ls file:///tmp/\n",
    "```\n",
    "\n",
    "and compare the results to\n",
    "\n",
    "```bash\n",
    "ls -alh /tmp/\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "none of this is to say *exclusively prefer* `hadoop fs` or *avoid* `hdfs dfs`. just knowing what the difference is may help you avoid some confusion when you try the subcommands or flags of one and don't experience the same result as you would with the other."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## web interfaces to `hadoop` services\n",
    "\n",
    "many of the `hadoop` services available via command line actions executed on our `ec2` instances are also available via web interfaces. as a matter of security, `aws` will only allow them to be accessed locally from that node. our best option is to use **`ssh` port forwarding** --  effectively, we are just replacing a port on our local machine with a single port on the remote machine.\n",
    "\n",
    "```bash\n",
    "ssh -i /path/to/your/keypair.pam -NfL $LOCAL_PORT:$REMOTE_PORT hadoop@ec2-###-###-###-###.compute-1.amazonaws.com\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">set up `ssh` port forwarding</div>**\n",
    "\n",
    "*windows users: check out [this walkthrough](https://www.akadia.com/services/ssh_putty.html)*\n",
    "\n",
    "+ you are running these commands on *your laptop*, not the `ec2` instance\n",
    "+ verify you can make an `ssh` connection\n",
    "    + you did this in the \"logging in to your emr cluster\" exercise above\n",
    "+ set the values of `MASTER_DNS` and `KEY_PAIR` below in a `bash` session **on your local laptop**\n",
    "\n",
    "```sh\n",
    "MASTER_DNS=YOUR_MASTER_DNS  # e.g.: ec2-3-229-137-205.compute-1.amazonaws.com\n",
    "KEY_PAIR=YOUR_KEY_PAIR      # e.g.: ~/.ssh/my_special_key.pem\n",
    "LOCAL_YARN=8088\n",
    "LOCAL_HDFS=50070\n",
    "LOCAL_SPARK=18080\n",
    "LOCAL_HUE=8887\n",
    "\n",
    "# YARN ResourceManager\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_YARN:$MASTER_DNS:8088 hadoop@$MASTER_DNS\n",
    "\n",
    "# Hadoop HDFS NameNode\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_HDFS:$MASTER_DNS:50070 hadoop@$MASTER_DNS\n",
    "\n",
    "# Spark HistoryServer\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_SPARK:$MASTER_DNS:18080 hadoop@$MASTER_DNS\n",
    "\n",
    "# Hue\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_HUE:$MASTER_DNS:8888 hadoop@$MASTER_DNS\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "after the above we have\n",
    "\n",
    "+ `YARN ResourceManager`: [http://localhost:8088](http://localhost:8088)\n",
    "+ `Hadoop HDFS NameNode`: [http://localhost:50070](http://localhost:50070)\n",
    "+ `Spark HistoryServer`: [http://localhost:18080](http://localhost:18080)\n",
    "+ `Hue`: [http://localhost:8887](http://localhost:8887)\n",
    "    + note: I used local port 8887 to avoid my `jupyter` server (ip is reserved), but you don't have to"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "**<div align=\"center\">optional: set up `ssh` port forwarding for non-master-node services</div>**\n",
    "\n",
    "+ go to emr web console > cluster page > summary tab\n",
    "    + click on the security group link for \"Security groups for Core & Task\"\n",
    "    + make sure your current IP address has `ssh` access to it (security group settings, people!)\n",
    "+ go to emr web console > cluster page > hardware tab\n",
    "    + scroll in the table to the two public dns names\n",
    "    + verify `ssh` works with a simple `ssh -i /path/to/key.pem hadoop@hadoop@ec2-###-##-##-###.compute-1.amazonaws.com`\n",
    "+ replace the `CORE_DNS_1`, `CORE_DNS_2`, and `KEY_PAIR` values below with values of your choosing and run this **on your local laptop**\n",
    "\n",
    "```sh\n",
    "CORE_DNS_1=YOUR_CORE_DNS_1\n",
    "CORE_DNS_2=YOUR_CORE_DNS_2\n",
    "KEY_PAIR=YOUR_KEY_PAIR\n",
    "LOCAL_YARN_1=8042\n",
    "LOCAL_YARN_2=8043\n",
    "LOCAL_HDFS_1=50075\n",
    "LOCAL_HDFS_2=50076\n",
    "\n",
    "# YARN NodeManager\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_YARN_1:$CORE_DNS_1:8042 hadoop@$CORE_DNS_1\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_YARN_2:$CORE_DNS_2:8042 hadoop@$CORE_DNS_2\n",
    "\n",
    "# Hadoop HDFS DataNode\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_HDFS_1:$CORE_DNS_1:50075 hadoop@$CORE_DNS_1\n",
    "ssh -i $KEY_PAIR -NfL $LOCAL_HDFS_2:$CORE_DNS_2:50075 hadoop@$CORE_DNS_2\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "after the above we have\n",
    "\n",
    "+ `YARN NodeManager` for core 1: [http://localhost:8042](http://localhost:8042)\n",
    "+ `YARN NodeManager` for core 2: [http://localhost:8043](http://localhost:8043)\n",
    "+ `Hadoop HDFS DataNode` for core 1: [http://localhost:50075](http://localhost:50075)\n",
    "+ `Hadoop HDFS DataNode` for core 2: [http://localhost:50076](http://localhost:50076)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### the `hdfs` `NameNode` webapp\n",
    "\n",
    "for starters, we have a web interface for exploring the `hdfs` directory structure: http://localhost:50070/explorer.html#/. there isn't much to see at this point, but this is one alternative interface to the `hadoop` `cli`\n",
    "\n",
    "for example, go here: http://localhost:50070/explorer.html#/data to see the `shakespeare.txt` file we `put` into `hdfs` earlier this lecture"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `hue`\n",
    "\n",
    "the `h`adoop `u`ser `e`xperience -- or `hue` for short -- is an extremely useful web interface. this is a sort of one-stop-shop for interacting with several of the most prominent `hadoop` tools -- especially for defining and organizing jobs.\n",
    "\n",
    "after the port forwarding steps above, `hue` should be accessible at http://localhost:8887 or http://localhost:8888, depending on your choice of `LOCAL_HUE` port"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">`hue`-based `emr`-walkthrough</div>**\n",
    "\n",
    "+ first, create an admin login. ***USE USERNAME `hadoop`***. use whatever password you want, but please use username `hadoop`!!\n",
    "+ do the simple guided walkthrough `hue` provides\n",
    "+ talk about the different utilities together"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">exercise: actually using `hue` to interact with `hadoop`</div>**\n",
    "\n",
    "let's put some data out there. go to the `hue` file browser\n",
    "\n",
    "+ textcorpus\n",
    "    + on your laptop: `wget https://github.com/bbengfort/hadoop-fundamentals/blob/master/data/textcorpus.zip?raw=true -O textcorpus.zip`\n",
    "    + in `hue`, click on the \"Files\" icon on the left panel menu\n",
    "    + click the \"upload\" button and upload `textcorpus.zip`\n",
    "    + click the uploaded item and click the \"Extract\" button\n",
    "+ shakespeare\n",
    "    + on your laptop: `wget https://github.com/bbengfort/hadoop-fundamentals/blob/master/data/shakespeare.txt.zip?raw=true -O shakespeare.txt.zip`\n",
    "    + repeat the above\n",
    "+ python data\n",
    "    + on your laptop: `wget https://s3.amazonaws.com/hadoop.rzl.gu511.com/pyq_stripped.tar.gz`\n",
    "    + repeat (note: this may take a while!)\n",
    "+ note: you may get an error or two during extract, ignore"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "returning now to the **editor** (the `</>` icon in the top left) -- this app provides a single interface to edit different queries in `hadoop`-relevant languages:\n",
    "\n",
    "+ `sql`-like query commands: `hive`, `sparksql`, generic databases (`mysql`, `postgresql`, etc)\n",
    "+ `hadoop`-specific languages and tools: `pig`, `mapreduce`, `sqoop1`\n",
    "+ `spark` commands: `scala`, `spark`, and `pyspark`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "let's try using it to run a simple `pyspark` snippet against our `pyq_stripped` dataset. don't worry about not knowing `pyspark` just yet -- we just want to demonstrate it's possible\n",
    "\n",
    "```python\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "answers_file_name = 'hdfs:///user/hadoop/pyq_stripped/Answers.csv'\n",
    "\n",
    "z = spark.read.csv(answers_file_name, header=True)\n",
    "\n",
    "z.agg(F.avg('Score')).show()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">PAUSE FOR ZOOM BREAK</div>**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## working with distributed computing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "as we said above, `yarn` is the main resource manager and one of the main access points for computation. in the original instance of `hadoop`, however, the computational framework was a software called `mapreduce`\n",
    "\n",
    "knowing what `mapreduce` is helps illuminate the engineering paradigm at play in `hadoop` programs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `mapreduce`: a functional programming model\n",
    "\n",
    "`mapreduce` was [first proposed](https://research.google.com/archive/mapreduce.html) by google developers as a way of performing easily distributable computations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the name comes from the two \"pieces\":\n",
    "\n",
    "+ a `map` function takes input as a series of key-value pairs (\"kvps\") and performs the same computation on each pair, generating a (possibly empty) sequence of intermediate kvps\n",
    "    + this is often where analysis happens (usually)\n",
    "    + e.g. filter: take a key, check if it belongs in a list of acceptable keys, emit the kvp if yes, pass silently if no\n",
    "+ a `reduce` function takes a key and an iterator of values and process the values, usually to determine some aggregate statistic\n",
    "\n",
    "these functions ought to be stateless functional programming functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "take this diagram representing a simple map-reduce pair which, end-to-end, make a word count job\n",
    "\n",
    "+ the `map` step takes a single input row (`A B R`) and emits a list of kvps which are the elements of the row and the number `1` (`A, 1`, `B, 1`, `R, 1`)\n",
    "+ the `reduce` step takes a list of `?, 1` values and *reduces* it by counting the `1`s (summing them)\n",
    "\n",
    "<img src=\"https://www.tutorialspoint.com/map_reduce/images/mapreduce_work.jpg\" width=\"700px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### `mapreduce`: implemented on a cluster\n",
    "\n",
    "the `mapreduce` framework is great for a distributed computation environment because it is assumes many of the central tenets of the distribution framework. specifically, because mappers and reducers are stateless functions, they can be executed by a worker node to independently work on any number of blocks and emit their responses back to the master node."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "mappers are already set: individual blocks are key-value pairs where the keys are file or line metadata and the values are the contents of the file / line. we can distribute the mapper function to any number of workers and let them process blocks at their own pace without any outside information\n",
    "\n",
    "did the machine `map`-ping `A B R` die? just ask the other machine with that file block to process it!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "reducers needs all the output values for a single key across all processed blocks, so we have to wait until all mappers are done to \"reduce\".\n",
    "\n",
    "we create as many reducers as there are output keys and distribute them among the workers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "because reducers expect to get the keys emitted by mappers and **all** values for those keys, we need to perform a shuffle and sort of those intermediate kvps before we can reduce. this stage is called exactly that: *shuffle and sort*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "so, in the end, we have a general framework:\n",
    "\n",
    "+ input: `hdfs` kvps\n",
    "+ mapping: input kvps are processed by mappers and generate intermediate kvps\n",
    "+ shuffle and sort: take the generated key, partition the key space, and assign keys to reducers\n",
    "+ reduce: take the keys and the iterated list of values and reduce them to aggregate kvps\n",
    "\n",
    "it's kvps all the way down!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "##### mapreduce examples\n",
    "\n",
    "we already counted words in the shakespeare corpus, in memory in plain `python`, and the pseudo-code which can fit this wordcount problem into `mapreduce` is not that different:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "```python\n",
    "def mapper(documentkey, line):\n",
    "    for word in line.split():\n",
    "        emit(word, 1)\n",
    "\n",
    "def reducer(word, values):\n",
    "    emit(word, sum(val for val in values))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "#### submitting a mapreduce job to `yarn`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`yarn` is responsible for scheduling tasks, so if we would like to perform some task we need to give it to `yarn`.\n",
    "\n",
    "one way (and the most basic) is to create a `jar` file (compiled `java` code) and to pass that directly to `yarn` using the `hadoop jar` command."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "in [the github repo](https://github.com/bbengfort/hadoop-fundamentals) for the \"Data Analytics with Hadoop\" O'Reilly book, we have been provided with a couple `java` files to implement a simple `mapreduce` word count job\n",
    "\n",
    "+ [`WordCount.java`](https://github.com/bbengfort/hadoop-fundamentals/blob/master/wordcount/WordCount/WordCount.java)\n",
    "+ [`WordMapper.java`](https://github.com/bbengfort/hadoop-fundamentals/blob/master/wordcount/WordCount/WordMapper.java)\n",
    "+ [`SumReducer.java`](https://github.com/bbengfort/hadoop-fundamentals/blob/master/wordcount/WordCount/SumReducer.java)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "let's compile and run that code on the shakespeare corpus."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "first thing's first, let's compile our `java` code into a `jar` file (run this on your master node)\n",
    "\n",
    "```bash\n",
    "export HADOOP_CLASSPATH=$JAVA_HOME/java/lib/tools.jar\n",
    "cd ~/code/hadoop-fundamentals/wordcount/WordCount/\n",
    "javac *.java -cp $(hadoop classpath)\n",
    "jar cf wc.jar WordCount.class WordMapper.class SumReducer.class\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "final thing's final, we can submit the `jar` file to `yarn` by calling\n",
    "\n",
    "```bash\n",
    "# use hadoop to submit a jar\n",
    "# that jar is wc.jar\n",
    "# the class to invoke is WordCount\n",
    "# the input is a file /data/shakespeare.txt\n",
    "# the output files go in a direcotry named wordcounts\n",
    "hadoop jar wc.jar WordCount /data/shakespeare.txt wordcounts\n",
    "```\n",
    "\n",
    "we can track the results of that job via a web interface at http://localhost:8088"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the output is now saved in `hdfs`:\n",
    "\n",
    "```sh\n",
    "hadoop fs -ls wordcounts/\n",
    "hadoop fs -cat wordcounts/part-r-00000 | head -n20\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">PAUSE FOR ZOOM BREAK</div>**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### not using `java`  with `hadoop` streaming"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "so we were able to take someone else's writen `java` code to create `mapreduce` jobs. super.\n",
    "\n",
    "I mean... not knowing `java` is a bit of a problem though. not to be ungrateful."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "this is just what we get out of the box with `hadoop` `mapreduce`.\n",
    "\n",
    "+ `java` `api` with input, output, map and reduce functions, job params exposed as *job configuration*\n",
    "+ jobs get packaged into a `jar` file which is passed to the `ResourceManager` by the *job client*\n",
    "+ `ResourceManager` handles the rest"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "but what if you don't want to write `java` code that implements this same workflow over and over and over again?\n",
    "\n",
    "or just don't want to write `java` code *at all*, because you already did everything you needed to do in `python`?\n",
    "\n",
    "*hadoop streaming* is here to help."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `hadoop` streaming\n",
    "\n",
    "hadoop streaming is a specific `java` util which can take any executable (in *any* language!) and use that as a mapper or reducer or combiner.\n",
    "\n",
    "really, this is just a super hacky `jar` file that is submitted in the same was as our `wc.jar` in our example above. for this `hadoop`-specific `jar` file, you pass executable scripts or commands as parameters to this `jar` file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "note: the word \"streaming\" is used because the input and output method is unix streams (`stdin`, `stdout`), not in reference to streaming data.\n",
    "\n",
    "this is actually pretty cool, because we know how to access those streams:\n",
    "\n",
    "+ `python`: `sys` module\n",
    "+ `R`: `file(\"stdin\")` (I think? who even knows. does anyone?)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "when we develop a `mapper.py` script, know the following:\n",
    "\n",
    "+ *each* mapper launches the executable. long spin-up times suck for obvious reasons\n",
    "+ `hadoop streaming` parses input data into lines of text and pipes them through `stdin`\n",
    "+ `python` streaming script parses those lines of texts and prints (to `stdout`) kvps delimited in some way (default is `\\t`)\n",
    "+ these intermediate kvps are scooped up by `hadoop streaming` again and passed on to the reducer\n",
    "+ the mapper gets an entire block via `sys.stdin`. so it doesn't receive a *file*, or a *line number*, it receives a file handler to a block. that's important."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the `reducer.py` script follows much of the same logic, but in addition:\n",
    "\n",
    "+ the reducer doesn't receive a key and an iterable, it reads shuffled and sorted kvp records (like a table) from stdin (they are in the `a\\tb` format)\n",
    "+ a single reducer task will always get *all* records for given key, but *may* get more than one key (so your reducer doesn't have a key, we need logic there)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "for both files (and for any file in any language being used as a `hadoop streaming` script), the shebang (`#!`) declaration at the top of the file is important -- it tells the streaming process (a bash shell) how to execute the script (e.g. in `python`)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### real example: flight data\n",
    "\n",
    "the [bureau of transportation statistics](https://transtats.bts.gov/) makes [on-time flight data](https://www.transtats.bts.gov/DL_SelectFields.asp?Table_ID=236&DB_Short_Name=On-Time) publicly available.\n",
    "\n",
    "let's download some and use a pre-written `mapper.py` and `reducer.py` to calculate the average delay per airport"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "first, on your `emr` master node, download the airline data:\n",
    "\n",
    "```bash\n",
    "cd /tmp\n",
    "wget --no-check-certificate -O flights.zip https://transtats.bts.gov/PREZIP/On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2020_1.zip\n",
    "unzip flights.zip\n",
    "# annoying file name!\n",
    "mv On_Time_Reporting_Carrier_On_Time_Performance_\\(1987_present\\)_2020_1.csv flights.csv\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the code we will use is in\n",
    "\n",
    "```bash\n",
    "cd ~/code/hadoop-fundamentals/avgdelay\n",
    "```\n",
    "\n",
    "let's take a look at `mapper.py` and `reducer.py` in that repo."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "one nice thing about this simple framework is that we can test our functions in a simple series of pipes:\n",
    "\n",
    "```bash\n",
    "head -n100 /tmp/flights.csv | ./mapper.py | sort | ./reducer.py\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "so, we have seen that the `avgdelay` code is able to `map` and `reduce` the records in the `csv` of airport delays we downloaded. let's ship that over to `hdfs` and run a `streaming` job with these files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "```bash\n",
    "hadoop fs -put /tmp/flights.csv .\n",
    "hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar \\\n",
    "    -files mapper.py,reducer.py \\\n",
    "    -input flights.csv \\\n",
    "    -output average_delay \\\n",
    "    -mapper mapper.py \\\n",
    "    -reducer reducer.py\n",
    "\n",
    "# trust, but verify\n",
    "hadoop fs -cat average_delay/part* | head -n25\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "note the `-files` params -- what's going on there?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "in a cluster environment, the materials of the executed code will generally have to be either\n",
    "\n",
    "1. pre-installed on the cluster, so that it is obvious\n",
    "2. shipped along with the request to execute."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">PAUSE FOR ZOOM BREAK</div>**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## the limitations of `hadoop`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "so now that we've spent a bunch of time on `hadoop`, it's time to come clean:\n",
    "\n",
    "`hadoop` is pretty hard to use (do *you* want to write `mapper.py` and `reducer.py` jobs for everything you do?), and is becoming less useful every day as other technologies - especially distributed database and `spark` -- gain popularity and efficiency."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "particularly for iterative data science workflows, `pyspark` is *much* closer to the ideal workflow. `hadoop` is very specifically geared for `etl` processes -- files in, files out. if you want to do repeat iteration on objects in an interactive environment, `hadoop` is very much not for you!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "the important thing to know about `hadoop`, in the end, is that it has some use cases for which it is very legitimately **the** solution to the problem (maybe only, but definitely standard). that being said, it is often **a pretty bad** solution to a problem, but one that the people who came before you suggested for... reasons"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "if you are thinking of using `hadoop` (or being told to use `hadoop`), ask yourself\n",
    "\n",
    "+ do I have a fixed input and a fixed output?\n",
    "+ is this a single-shot process that I want to repeat?\n",
    "+ am I done with exploration?\n",
    "\n",
    "if so, `hadoop` may still be a good choice. if not, you probably want `spark` -- or something else entirely"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## software lightning round"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `hadoop` and `spark`\n",
    "\n",
    "the current leaders in the distributed data and distributed computation sphere.\n",
    "\n",
    "`hadoop` is the on-disk distributed file processing tool which we've talked about here\n",
    "\n",
    "`spark` is the in-memory analog of `hadoop` and we will discuss this at length in the following lecture"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `pig`\n",
    "\n",
    "[`pig`](https://pig.apache.org/) is a declarative querying / scripting language (not like `sql`, more like `spark`). basically this is a higher-order abstraction of `mapreduce` functions that can be extremely fast while also being much easier to read and write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `hive`\n",
    "\n",
    "[`hive`](https://hive.apache.org/) is the `sql` of the `hadoop` ecosystem. it also doubles as a data warehousing platform for many of the other applications. it implements batch querying (not interactive querying), and is not *itself* a database, just a means of interacting with other data structures\n",
    "\n",
    "I personally find that `hive` is the easiest of the `hadoop` ecosystem tools to spin up in (especially if you have a background in `sql`), so you may want to consider staring here."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `hcatalog`\n",
    "[`hcatalog`](https://cwiki.apache.org/confluence/display/Hive/HCatalog) is a sub-component of `hive` that sees wide use in other applications. it is the table and storage management layer, and is used by disparate apps to provide a normalized, gridded view of data of many different formats. it acts as a normalized layer between application and data format"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `oozie`\n",
    "\n",
    "[`oozie`](http://oozie.apache.org/) is the scheduling and workflow management tool for `hadoop` jobs. it uses `xml` configuration files to define chained DAGs of jobs, executes them, reports on results and logs, handles concurrency and timing issues, and provides for a complex set of control flows.\n",
    "\n",
    "personally I think it's pretty janky, but its' the best you've got."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `hue`\n",
    "\n",
    "[`hue`](http://gethue.com/) (short for `h`adoop `u`ser `e`xperience) aims to be the way that users experience `hadoop`, and that's a great thing -- almost everything you might want to do in the `hadoop` ecosystem will be easier (at first) to do in `hue`.\n",
    "\n",
    "over time, of course, complications and bugs will push you down (up?) into the command line and the programs themselves, but for beginning it is *essential* that you try using `hue`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**<div align=\"center\">quick `hive` example using `hue`</div>**\n",
    "\n",
    "+ open `hue` at `http://localhost:8887` or `http://localhost:8888`\n",
    "    + earlier we should have uploaded and extracted a set of `python` questions from stack overflow\n",
    "+ create new database `stack_overflow_python`\n",
    "    + click the table icon on the left > click the `+` icon > follow the wizard for each of the three unzipped files\n",
    "    + note: creating a table **removes** the `csv` from the original `hdfs` location\n",
    "+ do some simple queries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `mahout`\n",
    "\n",
    "[`mahout`](http://mahout.apache.org/) is a framework for building machine learning applications. it is not immediately clear to me the extent to which `mahout` and `spark`'s `mllib` are *competitors* or *complimentors*, and I haven't been able to clear that up."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `sqoop`\n",
    "\n",
    "[`sqoop`](http://sqoop.apache.org/) (`sq`l `o`n hado`op`) is a bulk data transfer took, looking to implement (under the hood) `mapreduce` jobs to perform data transfer between traditional databases, `hdfs`, and other data sources."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `tez`\n",
    "[`tez`](https://tez.apache.org/) is similar to `spark` in that it is positioning itself as a replacement for `mapreduce` for simpler application frameworks. however, it is trying to be much closer to the `hadoop` / `yarn` infrastructure, implementing a normalized `api` so that other tools (the `pig`s and `hive`s of the world) can run directly off of `tez` in a normalized way."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `hbase`\n",
    "[`hbase`](https://hbase.apache.org/) is a big data database and date store. it is actually a `nosql` key-value store on `hdfs`, and it has its own query language (like many `nosql` dbs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `phoenix`\n",
    "[`phoenix`](https://phoenix.apache.org/)\n",
    "\n",
    "an on-line transaction processing software for performing data transactions in a `sql` syntax, but using `hbase` as its backing (so schemaless)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `presto`\n",
    "[`presto`](https://prestodb.io/) is a `sql` query engine for interactive queries against big data platforms. it was developed at facebook so it has good integration with `cassandra` in addition to other data stores.\n",
    "\n",
    "it's main goal is to be interactive and very fast. I haven't used it myself but the reviews are pretty positive."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `zeppelin`\n",
    "[`zeppelin`](https://zeppelin.apache.org/) is the `jupyter notebook` of the `spark` world, allowing for collaborative and exploratory work to be done in a web-based notebook. current interpreters include `spark`, `sql`, `python`, `hive`, `pig`, `sparksql`, `markdown`, and others"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `flink`\n",
    "[`flink`](https://flink.apache.org) is a stream processing framework. many of the big data sources of import are streams of data (log files, e.g.) and `flink` aims to be the standard means of persisting data streams into distributed environments"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `zookeeper`\n",
    "[`zookeeper`](https://zookeeper.apache.org/) is a centralized configuration engine. with so many moving parts in the distributed cluster environment, this can be pretty essential"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `livy`\n",
    "[`livy`](https://livy.incubator.apache.org/) is a `REST` interface for `spark`, allowing users to submit `spark` jobs from anywhere (not just on the cluster, in a `repl`)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `ganglia`\n",
    "[`ganglia`](http://ganglia.sourceforge.net/) is the primary distributed monitoring system for `hadoop` clusters and grid computing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### `mxnet`\n",
    "[`mxnet`](https://mxnet.incubator.apache.org/) is an apache incubator project for developing distributed deep learning algorithms"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## DELETE YOUR `emr` CLUSTER!\n",
    "\n",
    "don't forget to do it! if you are going to work on a homework assignment feel free to leave it up, but if you aren't *immediately* doing that, consider deleting it and cloning it when you are ready for your hw assignment"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "<strong><em><div align=\"center\">`hadoop` is `hadope`</div></em></strong>\n",
    "<div align=\"center\"><img src=\"https://pbs.twimg.com/media/Bnz0UglCUAA_Kb-.jpg\"></div>\n",
    "\n",
    "# END OF LECTURE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "## appendix\n",
    "\n",
    "the following are some extra items I thought would be useful to share but are not really a part of this lecture"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "### a development `hadoop` environment\n",
    "\n",
    "a single `hadoop` (or, related, `aws` `emr`) environment is often a large, complicated, expensive, and unruly engineering project.\n",
    "\n",
    "to avoid the hassle of constantly building up complicated development environments, many developers will create a *virtual execution environment* in either a *virtual machine* or a *`docker` container*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "#### using `virtualbox` and a pre-built vm image\n",
    "\n",
    "we are going to build one such virtual environment right now using oracle's `virtualbox` and the Ubuntu 14.04 `vmdk` provided by the authors of our text book"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "**<div align=\"center\">walkthrough: installing `virtualbox` and a `hadoop` virtual machine</div>**\n",
    "\n",
    "1. download https://resources.oreilly.com/examples/0636920035275/raw/master/hfpd3.vmdk.gz\n",
    "2. download `virtualbox` for your os and follow instructions: https://www.virtualbox.org/wiki/Downloads\n",
    "4. unzip the `vmdk` file once it is downloaded\n",
    "    1. in a terminal, `gunzip -k hfpd3.vmdk.gz`\n",
    "5. create the VM\n",
    "    1. open `virtualbox` and click the \"new\" button\n",
    "    2. name it whatever you want, change the type to `linux`, and make the version \"ubuntu 64-bit\"\n",
    "    3. set the memory however you want (I'll go high because yolo)\n",
    "    4. select \"Use an existing virtual hard disk file\" and navigate to the `vmdk` file\n",
    "    5. start up the VM. password is `password`\n",
    "    6. click \"Devices > Insert guest additions cd\" and the run (again, password is `password`)\n",
    "    7. restart the VM when finished (now you can resize!)\n",
    "    8. back in the `virtualbox` program, navigate to \"Settings\", and on the \"General > Advanced\" tab make \"Shared Clipboard\" \"Bidirectional\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "**<div align=\"center\">starting `hadoop`</div>**\n",
    "\n",
    "1. log in to your `hadoop` vm\n",
    "2. execute the following:\n",
    "\n",
    "```bash\n",
    "sudo -H -u hadoop $HADOOP_HOME/sbin/start-dfs.sh\n",
    "sudo -H -u hadoop $HADOOP_HOME/sbin/start-yarn.sh\n",
    "sudo -H -u hadoop $HADOOP_HOME/bin/hadoop fs -chown -R hadoop:hadoop /\n",
    "sudo -H -u hadoop $HADOOP_HOME/bin/hadoop fs -chown -R student:student /user/student\n",
    "sudo -H -u hadoop $HADOOP_HOME/bin/hadoop fs -chmod g+w /\n",
    "sudo chmod g+w /var/app/hadoop/data\n",
    "\n",
    "# demonstrate it worked\n",
    "hadoop fs -mkdir -p /user/student\n",
    "hadoop fs -ls -h /\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "#### using a `cloduera` `docker` `image`\n",
    "\n",
    "there are a few publicly available `docker` `image`s distributed by `cloudera`. several older versions exist on `dockerhub` and have an incredibly low barrier to entry as a result of this.\n",
    "\n",
    "the more up-to-date versions can be found on [the `cloudera` quickstart page](https://www.cloudera.com/downloads/quickstart_vms/5-13.html), and though they are also easy to install, it requires some identifying information to sign up and I try to limit that when possible.\n",
    "\n",
    "both images are pretty excellent and well documented so I highly recommend them for your general development and hacking."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "for both of the `docker` images below, you should increase your vm memory as discussed [here for mac / windows](https://stackoverflow.com/questions/44533319/how-to-assign-more-memory-to-docker-container). I recommend `8 GB`\n",
    "\n",
    "for linux, just increase memory by invoking `docker run` (when you do) with the command line flag `--memory=8g`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "##### cloudera 5.7\n",
    "\n",
    "this version is freely available on `dockerhub` and accessible via\n",
    "\n",
    "```sh\n",
    "docker pull cloudera/quickstart:latest\n",
    "```\n",
    "\n",
    "##### cloudera 5.13\n",
    "\n",
    "back on [the `cloudera` quickstart page](https://www.cloudera.com/downloads/quickstart_vms/5-13.html) you have the option to sign up and download a cloudera quickstart `docker` `image` for the most recent version of the cloudera distribution. if you choose to do this, installation instructions can be found [here]().\n",
    "\n",
    "for me, on my mac, the following are sufficient:\n",
    "\n",
    "```sh\n",
    "tar -xzvf cloudera-quickstart-vm-5.13.0-0-beta-docker.tar.gz\n",
    "cd cloudera-quickstart-vm-5.13.0-0-beta-docker\n",
    "docker import cloudera-quickstart-vm-5.13.0-0-beta-docker.tar\n",
    "```\n",
    "\n",
    "this takes a couple of minutes even if it's working as planned. when it finishes it will print out a `sha` (for example, mine was `sha256:e7701e8fac26ab951ee2b404594c43cac7d5e8f00b6b1d1dac73f1d6ede48904`). copy that value and add a tag:\n",
    "\n",
    "```sh\n",
    "docker tag e7701e8fac26ab951ee2b404594c43cac7d5e8f00b6b1d1dac73f1d6ede48904 cloudera/quickstart:5.13\n",
    "docker tag e7701e8fac26ab951ee2b404594c43cac7d5e8f00b6b1d1dac73f1d6ede48904 cloudera/quickstart:latest\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "##### running your `cloudera` `docker` `image`\n",
    "\n",
    "whichever `image` you acquired above, you can run it with\n",
    "\n",
    "```sh\n",
    "docker run \\\n",
    "    --hostname=quickstart.cloudera \\\n",
    "    --privileged=true \\\n",
    "    -rm \\\n",
    "    -it \\\n",
    "    -p $HOST_PORT_HUE:8888 \\\n",
    "    -p $HOST_PORT_CLOUDERA_MANAGER:7180 \\\n",
    "    -p $HOST_PORT_TUTORIAL:80 \\\n",
    "    $CLOUDERA_QUICKSTART_IMAGE_NAME\n",
    "    /usr/bin/docker-quickstart\n",
    "```\n",
    "\n",
    "for the `$CLOUDERA_QUICKSTART_IMAGE_NAME` value, you are looking for the name of the `image` you just created. you can find this from `docker images`\n",
    "\n",
    "for the 5.7 `image`, this is certainly `cloudera/quickstart:latest`. for the 5.13 `image`, this is whatever tag values you entered above (e.g. `cloudera/quickstart:5.13\n",
    "\n",
    "for example, the following oneliner has my preferred ports (I didn't want to use 8888 (`jupyter` was running) or `80` (I have a web server running):\n",
    "\n",
    "```sh\n",
    "docker run --hostname=quickstart.cloudera --privileged=true --rm -it -p 9999:8888 -p 7180:7180 -p 8880:80 cloudera/quickstart:latest /usr/bin/docker-quickstart\n",
    "```\n",
    "\n",
    "in the 5.7 `image`, `hue` *appears* to fail to start this way, but in my experience it succeeds (on a delay of a few seconds). if not, instructions on how to get that to work are [here](https://medium.com/@SnazzyHam/how-to-get-up-and-running-with-clouderas-quickstart-docker-container-732c04ed0280), but basically you will just run\n",
    "\n",
    "```sh\n",
    "service hue start\n",
    "```\n",
    "\n",
    "inside the `container` and check again.\n",
    "\n",
    "for what it's worth, a `cloudera` proprietary tool called cloudera manager isn't started by default. ***it recommends the VM have at least 8 GB memory to run this (we set 8 above)***. if you want to run (not recommended at this time), execute\n",
    "\n",
    "```sh\n",
    "/home/cloudera/cloudera-manager --express --force\n",
    "```\n",
    "\n",
    "in addition to the command line prompt the above `docker run` command will drop you at, this will create several links to which you may now navigate:\n",
    "\n",
    "+ [hue](http://localhost:9999)\n",
    "    + username and password: `cloudera`\n",
    "+ [cloudera manager](http://localhost:7180)\n",
    "+ [tutorial](http://localhost:8880)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "#### `cloudera` version comparison\n",
    "\n",
    "| observable     | 5.6            | 5.13            |\n",
    "|----------------|----------------|-----------------|\n",
    "| java version   | 1.7.0_67       | 1.7.0_67        |\n",
    "| hadoop version | 2.6.0-cdh5.7.0 | 2.6.0-cdh5.13.0 |\n",
    "| spark version  | 1.6.0          | 1.6.0           |\n",
    "| hive version   | x.y.z          | 1.1.0           |\n",
    "| hue version    | 3.9.0          | 3.9.0           |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "notes"
    }
   },
   "source": [
    "useful links\n",
    "\n",
    "+ [quora article on distribution and rack awareness](https://www.quora.com/How-does-HDFS-split-files)\n",
    "+ [great `hdfs` tutorial](https://www.edureka.co/blog/apache-hadoop-hdfs-architecture/?utm_source=quora&utm_medium=crosspost&utm_campaign=social-media-edureka-ab)\n",
    "+ [seemingly up-to-date table of `hadoop` ecosystem components](https://hadoopecosystemtable.github.io/)\n",
    "+ [deeper dive on small file size impacts](https://community.hitachivantara.com/community/products-and-solutions/pentaho/blog/2017/12/15/working-with-small-files-in-hadoop-part-1)"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}