065-etl-map-reduce.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "unset GZIP"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Extract Transform Load, Map Reduce"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ETL, MapReduce\n",
    "\n",
    "- often data needs to be preprocessed prior to training\n",
    "- common frameworks:\n",
    "    - `TFRecord` + Google Map Reduce\n",
    "    - Hadoop, Spark (Sequence files, Parquet, Avro)\n",
    "- Issues\n",
    "    - not K8s native\n",
    "    - use serialized data structures rather than files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# ETL/MapReduce based on Files\n",
    "\n",
    "Basic approach:\n",
    "- files + sequential storage = `tar` archives\n",
    "- same format as `WebDataset`\n",
    "- process each shard with `tarproc`\n",
    "\n",
    "Big datasets:\n",
    "- split a single dataset into multiple archives to enable more parallelism\n",
    "- store in object stores (Swift, AIStore), cloud storage buckets (S3, Azure, Google)\n",
    "- process shards in parallel with K8s Jobs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# ImageNet Bucket\n",
    "\n",
    "We have a bucket consisting of ImageNet shards that we want to perform format conversions on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "gs://lpr-imagenet/imagenet_train-0000.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0001.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0002.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0003.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0004.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0005.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0006.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0007.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0008.tgz\n",
      "gs://lpr-imagenet/imagenet_train-0009.tgz\n",
      "close failed in file object destructor:\n",
      "sys.excepthook is missing\n",
      "lost sys.stderr\n"
     ]
    }
   ],
   "source": [
    "gsutil ls gs://lpr-imagenet | sed 10q"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "gs://lpr-imagenet/imagenet_train-0000.tgz\n",
      "gs://lpr-imagenet/imagenet_val-0000.tgz\n"
     ]
    }
   ],
   "source": [
    "gsutil ls gs://lpr-imagenet | sed 's/[0-9]/0/g' | sort -u"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Shard Contents\n",
    "\n",
    "- groups of `.cls`, `.jpg`, and `.json` files\n",
    "- all files comprising a training sample have the same basename\n",
    "- everything past the first dot: `.input.png`, `.output.png`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-rw-rw-rw- bigdata/bigdata   3 2019-06-08 12:12 n03788365_17158.cls\n",
      "-rw-rw-rw- bigdata/bigdata 75884 2019-06-08 12:12 n03788365_17158.jpg\n",
      "-rw-rw-rw- bigdata/bigdata   382 2019-06-08 12:12 n03788365_17158.json\n",
      "-rw-rw-rw- bigdata/bigdata     3 2019-06-08 12:12 n03000247_49831.cls\n",
      "-rw-rw-rw- bigdata/bigdata 57068 2019-06-08 12:12 n03000247_49831.jpg\n",
      "-rw-rw-rw- bigdata/bigdata   104 2019-06-08 12:12 n03000247_49831.json\n",
      "-rw-rw-rw- bigdata/bigdata     3 2019-06-08 12:12 n03000247_22907.cls\n",
      "-rw-rw-rw- bigdata/bigdata 97447 2019-06-08 12:12 n03000247_22907.jpg\n",
      "-rw-rw-rw- bigdata/bigdata   450 2019-06-08 12:12 n03000247_22907.json\n",
      "-rw-rw-rw- bigdata/bigdata     3 2019-06-08 12:12 n04597913_10741.cls\n",
      "tar: write error\n"
     ]
    }
   ],
   "source": [
    "gsutil cat gs://lpr-imagenet/imagenet_train-0000.tgz | tar -ztvf - | sed 10q"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# The `tarproc` Command Processes Tar Archives\n",
    "\n",
    "- extract each group of files into a temporary directory\n",
    "- run the command\n",
    "- (optionally) package the result into a new tar file\n",
    "- delete the temporary directory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "__key__  __source__  sample.cls  sample.jpg  sample.json\n",
      "__key__  __source__  sample.cls  sample.jpg  sample.json\n",
      "__key__  __source__  sample.cls  sample.jpg  sample.json\n"
     ]
    }
   ],
   "source": [
    "gsutil cat gs://lpr-imagenet/imagenet_train-0000.tgz | tarproc -c ls --count 3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# The `__key__` Identifies Samples\n",
    "\n",
    "- extracted files always use `sample` as their basename\n",
    "- the `__key__` file contains the basename of the file inside the archive"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "n03788365_17158\n",
      "n03000247_49831\n",
      "n03000247_22907\n"
     ]
    }
   ],
   "source": [
    "gsutil cat gs://lpr-imagenet/imagenet_train-0000.tgz | tarproc -c 'cat __key__; echo' --count 3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Example: Large Scale Image Format Conversion\n",
    "\n",
    "- convert each `.jpg` image to `.ppm` (using ImageMagick)\n",
    "- store result in a new `.tar` file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "gsutil cat gs://lpr-imagenet/imagenet_train-0000.tgz |\n",
    "tarproc -c '\n",
    "    convert sample.jpg sample.ppm\n",
    "    rm sample.jpg\n",
    "' --count 3 -o out.tar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-r--r--r-- bigdata/bigdata   3 2019-12-09 15:24 n03788365_17158.cls\n",
      "-r--r--r-- bigdata/bigdata 382 2019-12-09 15:24 n03788365_17158.json\n",
      "-r--r--r-- bigdata/bigdata 478515 2019-12-09 15:24 n03788365_17158.ppm\n",
      "-r--r--r-- bigdata/bigdata      3 2019-12-09 15:24 n03000247_49831.cls\n",
      "-r--r--r-- bigdata/bigdata    104 2019-12-09 15:24 n03000247_49831.json\n",
      "-r--r--r-- bigdata/bigdata 499515 2019-12-09 15:24 n03000247_49831.ppm\n",
      "-r--r--r-- bigdata/bigdata      3 2019-12-09 15:24 n03000247_22907.cls\n",
      "-r--r--r-- bigdata/bigdata    450 2019-12-09 15:24 n03000247_22907.json\n",
      "-r--r--r-- bigdata/bigdata 562515 2019-12-09 15:24 n03000247_22907.ppm\n"
     ]
    }
   ],
   "source": [
    "tar tvf out.tar"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# `tarproc` supports Multicore Processing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Copying gs://lpr-imagenet/imagenet_train-0000.tgz...\n",
      "/ [1/1 files][946.3 MiB/946.3 MiB] 100% Done  82.4 MiB/s ETA 00:00:00           \n",
      "Operation completed over 1 objects/946.3 MiB.                                    \n"
     ]
    }
   ],
   "source": [
    "gsutil -m cp gs://lpr-imagenet/imagenet_train-0000.tgz imagenet_train-0000.tgz"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "real\t3m19.963s\n",
      "user\t2m9.339s\n",
      "sys\t1m9.149s\n"
     ]
    }
   ],
   "source": [
    "time /bin/bash -c '\n",
    "cat imagenet_train-0000.tgz | tarproc -c \"convert sample.jpg sample.ppm; rm sample.jpg\" -o out.tar\n",
    "'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To enable multicore processing, just use the `-p` flag."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "real\t0m52.049s\n",
      "user\t2m4.979s\n",
      "sys\t1m13.184s\n"
     ]
    }
   ],
   "source": [
    "time /bin/bash -c '\n",
    "cat imagenet_train-0000.tgz | tarproc -p 8 -c \"convert sample.jpg sample.ppm; rm sample.jpg\" -o out.tar\n",
    "'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Processing Using `WebDataset`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "cat > mapper.py <<'EOF'\n",
    "import sys, argparse\n",
    "from webdataset.dataset import WebDataset\n",
    "from webdataset.writer import TarWriter\n",
    "\n",
    "parser = argparse.ArgumentParser(\"convert jpg to png in imagenet-style databases\")\n",
    "parser.add_argument(\"--count\", type=int, default=9999999999)\n",
    "parser.add_argument(\"input\"); parser.add_argument(\"output\")\n",
    "args = parser.parse_args()\n",
    "\n",
    "sink = TarWriter(args.output)\n",
    "for i, (key, image, cls) in enumerate(WebDataset(args.input, extensions=\"__key__ jpg cls\")):\n",
    "    if i%1000==0: print(i, key, file=sys.stderr)\n",
    "    if i>=args.count: break\n",
    "    sink.write(dict(__key__=key, ppm=image[::2,::2,:], cls=cls ))\n",
    "sink.close()\n",
    "EOF"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Running the Job\n",
    "\n",
    "You have the full power of PyTorch available, including GPU processing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 n03788365_17158\n",
      "1000 n02965783_5956\n",
      "2000 n10565667_409\n"
     ]
    }
   ],
   "source": [
    "python3 mapper.py --count 2000 imagenet_train-0000.tgz out1.tar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "cat > kubetpl.yaml <<'EOF'\n",
    "image: gcr.io/research-191823/bigdata19\n",
    "memory: 4G\n",
    "cpu: 1\n",
    "app: bigdata19\n",
    "port:\n",
    "  - 7880\n",
    "EOF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Error from server (NotFound): jobs.batch \"job0000\" not found\n",
      "job.batch/job0000 created\n"
     ]
    }
   ],
   "source": [
    "kubectl delete job/job0000 || true\n",
    "kubetpl job -n job0000 -c '\n",
    "curl -s http://storage.googleapis.com/lpr-imagenet/imagenet_train-0000.tgz |\n",
    "tarproc --count 3 -c \"ls -l\"\n",
    "' | kubectl apply -f -"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "sleep 15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 92\n",
      "-rw-r--r-- 1 root root    15 Dec  9 23:29 __key__\n",
      "-rw-r--r-- 1 root root     1 Dec  9 23:29 __source__\n",
      "-rw-r--r-- 1 root root     3 Dec  9 23:29 sample.cls\n",
      "-rw-r--r-- 1 root root 75884 Dec  9 23:29 sample.jpg\n",
      "-rw-r--r-- 1 root root   382 Dec  9 23:29 sample.json\n",
      "total 72\n",
      "-rw-r--r-- 1 root root    15 Dec  9 23:29 __key__\n",
      "-rw-r--r-- 1 root root     1 Dec  9 23:29 __source__\n",
      "-rw-r--r-- 1 root root     3 Dec  9 23:29 sample.cls\n",
      "-rw-r--r-- 1 root root 57068 Dec  9 23:29 sample.jpg\n",
      "-rw-r--r-- 1 root root   104 Dec  9 23:29 sample.json\n",
      "total 112\n",
      "-rw-r--r-- 1 root root    15 Dec  9 23:29 __key__\n",
      "-rw-r--r-- 1 root root     1 Dec  9 23:29 __source__\n",
      "-rw-r--r-- 1 root root     3 Dec  9 23:29 sample.cls\n",
      "-rw-r--r-- 1 root root 97447 Dec  9 23:29 sample.jpg\n",
      "-rw-r--r-- 1 root root   450 Dec  9 23:29 sample.json\n"
     ]
    }
   ],
   "source": [
    "kubectl logs job/job0000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "job.batch \"job0000\" deleted\n",
      "job.batch/job0000 created\n"
     ]
    }
   ],
   "source": [
    "kubectl delete job/job0000 || true\n",
    "kubetpl job -n job0000 -c '\n",
    "curl -s http://storage.googleapis.com/lpr-imagenet/imagenet_train-0000.tgz |\n",
    "tarproc --count 3 -c \"gm convert sample.jpg sample.ppm; rm sample.jpg\" -o - |\n",
    "dd of=/dev/null # put a copy-to-destination here\n",
    "' | kubectl apply -f -"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "sleep 15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3040+0 records in\n",
      "3040+0 records out\n",
      "1556480 bytes (1.6 MB, 1.5 MiB) copied, 0.679939 s, 2.3 MB/s\n"
     ]
    }
   ],
   "source": [
    "kubectl logs job/job0000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "job.batch \"job0000\" deleted\n",
      "No resources found\n"
     ]
    }
   ],
   "source": [
    "kubectl delete jobs --all\n",
    "kubectl delete pods --all"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Processing All Shards in Parallel on K8s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "job.batch/job0138 created\n",
      "job.batch/job0139 created\n",
      "job.batch/job0140 created\n",
      "job.batch/job0141 created\n",
      "job.batch/job0142 created\n",
      "job.batch/job0143 created\n",
      "job.batch/job0144 created\n",
      "job.batch/job0145 created\n",
      "job.batch/job0146 created\n",
      "job.batch/job0147 created\n"
     ]
    }
   ],
   "source": [
    "for i in {0000..0147}; do\n",
    "kubetpl job -n job$i -c \"\n",
    "curl -s http://storage.googleapis.com/lpr-imagenet/imagenet_train-$i.tgz |\n",
    "tarproc -c 'gm convert sample.jpg sample.ppm; rm sample.jpg' -o - |\n",
    "dd of=/dev/null # put a copy-to-destination here\n",
    "\" | kubectl apply -f - >> _log\n",
    "done\n",
    "tail _log"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "sleep 60"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "    148 Running\n"
     ]
    }
   ],
   "source": [
    "kubectl get pods | sed 1d | awk '{print $3}' | sort | uniq -c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "kubectl delete jobs --all >> _log\n",
    "kubectl delete pods --all >> _log"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "pod \"job0137-wdshf\" deleted\n",
      "pod \"job0138-b6qmr\" deleted\n",
      "pod \"job0139-fzhjj\" deleted\n",
      "pod \"job0140-k6xhz\" deleted\n",
      "pod \"job0141-wvkcz\" deleted\n",
      "pod \"job0142-5cw7k\" deleted\n",
      "pod \"job0143-4vd5k\" deleted\n",
      "pod \"job0144-wnrcs\" deleted\n",
      "pod \"job0145-lwmwd\" deleted\n",
      "pod \"job0146-xqb45\" deleted\n"
     ]
    }
   ],
   "source": [
    "tail _log"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# ETL, Map Reduce\n",
    "\n",
    "- DL and AI require large scale data preprocessing\n",
    "- we use sequential file formats and sharding to make this efficient\n",
    "- several platforms to choose from: Hadoop, Spark\n",
    "- DL/AI data is usually file based, so examples use POSIX archives\n",
    "- parallelization using K8s Jobs"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Bash",
   "language": "bash",
   "name": "bash"
  },
  "language_info": {
   "codemirror_mode": "shell",
   "file_extension": ".sh",
   "mimetype": "text/x-sh",
   "name": "bash"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}