How to write results into HDFS - example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How to write results into HDFS\n",
    "\n",
    "After submitting a job, we will need to retrieve the result. This can be stored in HDFS or elsewhere. Depending on the output size this can be a convenient approach or not. If so, we will need to write it in some format in order to read it back afterwards.\n",
    "\n",
    "To produce this test I'm using Spark 1.5.1 (Pyspark 1.5.1) and ```spark-avro``` libraries loaded like this:\n",
    "\n",
    "```bash\n",
    "spark-submit --packages com.databricks:spark-avro_2.10:2.0.1 [...]\n",
    "```\n",
    "\n",
    "For Spark 1.3.0 use\n",
    "\n",
    "```bash\n",
    "spark-submit --packages com.databricks:spark-avro_2.10:1.0.0 [...]\n",
    "```\n",
    "Example with Spark 1.3.0 is provided in a separated file.\n",
    "\n",
    "Index:\n",
    "\n",
    "* [How to store aggregation results](#Aggregation-example)\n",
    " * [Example #1](#First-example)\n",
    " * [Example #2](#Second-example)\n",
    "* How to store selection results\n",
    " * Example #1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.context.SparkContext at 0x7f10d8032650>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# is SparkContext already loaded?\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.sql.context.HiveContext at 0x7f10c96dbe50>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Make sure you have a HiveContext\n",
    "sqlContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "u'1.5.1'"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Which is the version?\n",
    "sc.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# load a dataframe from Avro files\n",
    "df = sqlContext.read.format(\"com.databricks.spark.avro\").load(\"/cms/wmarchive/test/avro/2016/01/01/\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- PFNArrayRef: array (nullable = false)\n",
      " |    |-- element: string (containsNull = false)\n",
      " |-- task: string (nullable = false)\n",
      " |-- skippedFiles: array (nullable = false)\n",
      " |    |-- element: long (containsNull = false)\n",
      " |-- wmaid: string (nullable = false)\n",
      " |-- wmats: double (nullable = false)\n",
      " |-- fallbackFiles: array (nullable = false)\n",
      " |    |-- element: long (containsNull = false)\n",
      " |-- LFNArray: array (nullable = false)\n",
      " |    |-- element: string (containsNull = false)\n",
      " |-- meta_data: struct (nullable = false)\n",
      " |    |-- agent_ver: string (nullable = false)\n",
      " |    |-- fwjr_id: string (nullable = false)\n",
      " |    |-- host: string (nullable = false)\n",
      " |    |-- ts: long (nullable = false)\n",
      " |-- steps: array (nullable = false)\n",
      " |    |-- element: struct (containsNull = false)\n",
      " |    |    |-- status: long (nullable = false)\n",
      " |    |    |-- errors: array (nullable = false)\n",
      " |    |    |    |-- element: struct (containsNull = false)\n",
      " |    |    |    |    |-- type: string (nullable = false)\n",
      " |    |    |    |    |-- details: string (nullable = false)\n",
      " |    |    |    |    |-- exitCode: long (nullable = false)\n",
      " |    |    |-- name: string (nullable = false)\n",
      " |    |    |-- output: array (nullable = false)\n",
      " |    |    |    |-- element: struct (containsNull = false)\n",
      " |    |    |    |    |-- branch_hash: string (nullable = false)\n",
      " |    |    |    |    |-- guid: string (nullable = false)\n",
      " |    |    |    |    |-- size: long (nullable = false)\n",
      " |    |    |    |    |-- applicationName: string (nullable = false)\n",
      " |    |    |    |    |-- acquisitionEra: string (nullable = false)\n",
      " |    |    |    |    |-- applicationVersion: string (nullable = false)\n",
      " |    |    |    |    |-- inputPFNs: array (nullable = false)\n",
      " |    |    |    |    |    |-- element: long (containsNull = false)\n",
      " |    |    |    |    |-- configURL: string (nullable = false)\n",
      " |    |    |    |    |-- outputDataset: string (nullable = false)\n",
      " |    |    |    |    |-- location: string (nullable = false)\n",
      " |    |    |    |    |-- inputLFNs: array (nullable = false)\n",
      " |    |    |    |    |    |-- element: long (containsNull = false)\n",
      " |    |    |    |    |-- async_dest: string (nullable = false)\n",
      " |    |    |    |    |-- events: long (nullable = false)\n",
      " |    |    |    |    |-- merged: long (nullable = false)\n",
      " |    |    |    |    |-- validStatus: string (nullable = false)\n",
      " |    |    |    |    |-- adler32: string (nullable = false)\n",
      " |    |    |    |    |-- ouput_module_class: string (nullable = false)\n",
      " |    |    |    |    |-- globalTag: string (nullable = false)\n",
      " |    |    |    |    |-- catalog: string (nullable = false)\n",
      " |    |    |    |    |-- module_label: string (nullable = false)\n",
      " |    |    |    |    |-- cksum: string (nullable = false)\n",
      " |    |    |    |    |-- StageOutCommand: string (nullable = false)\n",
      " |    |    |    |    |-- outputPFNs: array (nullable = false)\n",
      " |    |    |    |    |    |-- element: long (containsNull = false)\n",
      " |    |    |    |    |-- inputDataset: string (nullable = false)\n",
      " |    |    |    |    |-- runs: array (nullable = false)\n",
      " |    |    |    |    |    |-- element: struct (containsNull = false)\n",
      " |    |    |    |    |    |    |-- runNumber: long (nullable = false)\n",
      " |    |    |    |    |    |    |-- lumis: array (nullable = false)\n",
      " |    |    |    |    |    |    |    |-- element: long (containsNull = false)\n",
      " |    |    |    |    |-- outputLFNs: array (nullable = false)\n",
      " |    |    |    |    |    |-- element: long (containsNull = false)\n",
      " |    |    |    |    |-- processingVer: long (nullable = false)\n",
      " |    |    |    |    |-- processingStr: string (nullable = false)\n",
      " |    |    |    |    |-- prep_id: string (nullable = false)\n",
      " |    |    |-- stop: long (nullable = false)\n",
      " |    |    |-- site: string (nullable = false)\n",
      " |    |    |-- start: long (nullable = false)\n",
      " |    |    |-- input: array (nullable = false)\n",
      " |    |    |    |-- element: struct (containsNull = false)\n",
      " |    |    |    |    |-- runs: array (nullable = false)\n",
      " |    |    |    |    |    |-- element: struct (containsNull = false)\n",
      " |    |    |    |    |    |    |-- runNumber: long (nullable = false)\n",
      " |    |    |    |    |    |    |-- lumis: array (nullable = false)\n",
      " |    |    |    |    |    |    |    |-- element: long (containsNull = false)\n",
      " |    |    |    |    |-- input_source_class: string (nullable = false)\n",
      " |    |    |    |    |-- input_type: string (nullable = false)\n",
      " |    |    |    |    |-- lfn: long (nullable = false)\n",
      " |    |    |    |    |-- pfn: long (nullable = false)\n",
      " |    |    |    |    |-- catalog: string (nullable = false)\n",
      " |    |    |    |    |-- module_label: string (nullable = false)\n",
      " |    |    |    |    |-- guid: string (nullable = false)\n",
      " |    |    |    |    |-- events: long (nullable = false)\n",
      " |    |    |-- performance: struct (nullable = false)\n",
      " |    |    |    |-- storage: struct (nullable = false)\n",
      " |    |    |    |    |-- writeTotalMB: double (nullable = false)\n",
      " |    |    |    |    |-- readPercentageOps: double (nullable = false)\n",
      " |    |    |    |    |-- readAveragekB: double (nullable = false)\n",
      " |    |    |    |    |-- readTotalMB: double (nullable = false)\n",
      " |    |    |    |    |-- readNumOps: double (nullable = false)\n",
      " |    |    |    |    |-- readCachePercentageOps: double (nullable = false)\n",
      " |    |    |    |    |-- readMBSec: double (nullable = false)\n",
      " |    |    |    |    |-- readMaxMSec: double (nullable = false)\n",
      " |    |    |    |    |-- readTotalSecs: double (nullable = false)\n",
      " |    |    |    |    |-- writeTotalSecs: double (nullable = false)\n",
      " |    |    |    |-- cpu: struct (nullable = false)\n",
      " |    |    |    |    |-- TotalJobCPU: double (nullable = false)\n",
      " |    |    |    |    |-- AvgEventCPU: double (nullable = false)\n",
      " |    |    |    |    |-- MaxEventCPU: double (nullable = false)\n",
      " |    |    |    |    |-- AvgEventTime: double (nullable = false)\n",
      " |    |    |    |    |-- MinEventCPU: double (nullable = false)\n",
      " |    |    |    |    |-- TotalEventCPU: long (nullable = false)\n",
      " |    |    |    |    |-- TotalJobTime: double (nullable = false)\n",
      " |    |    |    |    |-- MinEventTime: double (nullable = false)\n",
      " |    |    |    |    |-- MaxEventTime: double (nullable = false)\n",
      " |    |    |    |-- memory: struct (nullable = false)\n",
      " |    |    |    |    |-- PeakValueRss: double (nullable = false)\n",
      " |    |    |    |    |-- PeakValueVsize: double (nullable = false)\n",
      " |-- PFNArray: array (nullable = false)\n",
      " |    |-- element: string (containsNull = false)\n",
      " |-- LFNArrayRef: array (nullable = false)\n",
      " |    |-- element: string (containsNull = false)\n",
      " |-- stype: string (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 5 ms, sys: 3 ms, total: 8 ms\n",
      "Wall time: 14.8 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "200000"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aggregation examples\n",
    "\n",
    "### First example\n",
    "1) Aggregated sum of all ```steps.performance.cpu``` values. In this case the result is a single line that can be easily stored back in HDFS, also in a textual format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "aggregation1 = df.select(\"steps.performance.cpu\") \\\n",
    "    .rdd \\\n",
    "    .flatMap(lambda cpuArrayRows: cpuArrayRows[0]) \\\n",
    "    .map(lambda row: row.asDict()) \\\n",
    "    .flatMap(lambda rowDict: [(k,v) for k,v in rowDict.iteritems()]) \\\n",
    "    .reduceByKey(lambda x,y: x+y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 15 ms, sys: 6 ms, total: 21 ms\n",
      "Wall time: 19.6 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[('TotalJobCPU', 152097347.4204235),\n",
       " ('TotalJobTime', 1652265.155986641),\n",
       " ('MinEventCPU', 1349112.0),\n",
       " ('MinEventTime', 300451.93912530516),\n",
       " ('AvgEventCPU', 299587.125072929),\n",
       " ('MaxEventTime', 1656177.8017510779),\n",
       " ('TotalEventCPU', 151599144),\n",
       " ('AvgEventTime', 300340.72760207055),\n",
       " ('MaxEventCPU', 1351100.0)]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "aggregation1.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Store the file as a simple text file\n",
    "aggregation1.saveAsTextFile(\"wmarchive/test-plaintext-aggregation1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('TotalJobCPU', 152097347.4204235)\n",
      "('TotalJobTime', 1652265.155986641)\n",
      "('MinEventCPU', 1349112.0)\n",
      "('MinEventTime', 300451.93912530516)\n",
      "('AvgEventCPU', 299587.125072929)\n",
      "('MaxEventTime', 1656177.8017510779)\n",
      "('TotalEventCPU', 151599144)\n",
      "('AvgEventTime', 300340.72760207055)\n",
      "('MaxEventCPU', 1351100.0)\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "hadoop fs -text wmarchive/test-plaintext-aggregation1/*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "aggregated1DF = sqlContext.createDataFrame([{v[0]:v[1] for v in aggregation1.collect()}])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# saving in Json format\n",
    "aggregated1DF.toJSON().saveAsTextFile(\"wmarchive/test-json-aggregation1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\"AvgEventCPU\":299587.125072929,\"AvgEventTime\":300340.72760207055,\"MaxEventCPU\":1351100.0,\"MaxEventTime\":1656177.8017510779,\"MinEventCPU\":1349112.0,\"MinEventTime\":300451.93912530516,\"TotalEventCPU\":151599144,\"TotalJobCPU\":1.520973474204235E8,\"TotalJobTime\":1652265.155986641}\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "hadoop fs -text wmarchive/test-json-aggregation1/*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# how to write in Avro format\n",
    "aggregated1DF.write.format(\"com.databricks.spark.avro\").save(\"wmarchive/test-avro-aggregation1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\"AvgEventCPU\":{\"double\":299587.125072929},\"AvgEventTime\":{\"double\":300340.72760207055},\"MaxEventCPU\":{\"double\":1351100.0},\"MaxEventTime\":{\"double\":1656177.8017510779},\"MinEventCPU\":{\"double\":1349112.0},\"MinEventTime\":{\"double\":300451.93912530516},\"TotalEventCPU\":{\"long\":151599144},\"TotalJobCPU\":{\"double\":1.520973474204235E8},\"TotalJobTime\":{\"double\":1652265.155986641}}\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "hadoop fs -text wmarchive/test-avro-aggregation1/*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Second example\n",
    "\n",
    "2) Count which is the most popular host"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}