notebooks/ceph-example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Reading input from Ceph Object Store with Apache Spark on OpenShift\n",
    "\n",
    "In this demonstration we will load textual data from [Ceph](http://ceph.com/) using [S3 API](http://docs.ceph.com/docs/master/radosgw/s3/). There are two key pieces of information to get from this demonstration,\n",
    "\n",
    "0. First, loading of the S3 client libraries (hadoop-aws)\n",
    "1. Second, configuring the client with Ceph/S3 credentials.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Important - load the S3 client libraries\n",
    "\n",
    "This uses some Jupyter line magic to put **--packages** on the pyspark command line for the kernel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "env: PYSPARK_SUBMIT_ARGS=--packages=org.apache.hadoop:hadoop-aws:2.7.3 pyspark-shell\n"
     ]
    }
   ],
   "source": [
    "%set_env PYSPARK_SUBMIT_ARGS=--packages=org.apache.hadoop:hadoop-aws:2.7.3 pyspark-shell"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Setup a bucket\n",
    "\n",
    "Before create a job in our Spark cluster, let's create a bucket using S3 API:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Bucket radanalytics created!\n"
     ]
    }
   ],
   "source": [
    "import boto\n",
    "import boto.s3.connection\n",
    "\n",
    "access_key = '4LO0VJO8MEJ3OU98OR7L'\n",
    "secret_key = 'ulB5kdTTfGXG7AjDaLxcL5VHpSOXFBBLfNRSqduA'\n",
    "bucket_name = \"radanalytics\"\n",
    "\n",
    "ceph_host = '192.168.121.211' # Add your rgw0 Ceph host here\n",
    "ceph_port = 8080\n",
    "\n",
    "conn = boto.connect_s3(\n",
    "        aws_access_key_id = access_key,\n",
    "        aws_secret_access_key = secret_key,\n",
    "        host = ceph_host,\n",
    "        port = ceph_port,\n",
    "        is_secure=False,\n",
    "        calling_format = boto.s3.connection.OrdinaryCallingFormat()\n",
    "        )\n",
    "\n",
    "bucket = conn.create_bucket(bucket_name)\n",
    "\n",
    "print \"Bucket {} created!\".format(bucket_name)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Put a file in the Bucket"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Object spark-test added in bucket radanalytics!\n"
     ]
    }
   ],
   "source": [
    "from boto.s3.key import Key\n",
    "\n",
    "object_key = \"spark-test\"\n",
    "object_value = \"/usr/local/spark/README.md\"\n",
    "\n",
    "bucket = conn.get_bucket(bucket_name)\n",
    "\n",
    "k = Key(bucket)\n",
    "k.key = object_key\n",
    "k.set_contents_from_filename(object_value)\n",
    "\n",
    "print \"Object {} added in bucket {}!\".format(object_key, bucket_name)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## List contents in the bucket"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "spark-test\t3818\t2017-07-03T18:49:15.466Z\n"
     ]
    }
   ],
   "source": [
    "for key in bucket.list():\n",
    "    print \"{name}\\t{size}\\t{modified}\".format(\n",
    "        name = key.name,\n",
    "        size = key.size,\n",
    "        modified = key.last_modified,\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Setup your SparkSession as you normally would."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql import SparkSession, SQLContext\n",
    "\n",
    "spark = SparkSession.builder.master(\"spark://sparky:7077\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Important - configure the S3 client with your credentials\n",
    "Don't store your credentials in code, use [Secrets](https://kubernetes.io/docs/user-guide/secrets/). Do use [AWS IAM](https://aws.amazon.com/iam/) and credentials with only the capabilities needed for your application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "hadoopConf=spark.sparkContext._jsc.hadoopConfiguration()\n",
    "hadoopConf.set(\"fs.s3a.access.key\", \"4LO0VJO8MEJ3OU98OR7L\")\n",
    "hadoopConf.set(\"fs.s3a.secret.key\", \"ulB5kdTTfGXG7AjDaLxcL5VHpSOXFBBLfNRSqduA\")\n",
    "hadoopConf.set(\"fs.s3a.endpoint\", \"192.168.121.211:8080\")\n",
    "hadoopConf.set(\"fs.s3a.connection.ssl.enabled\", \"false\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "This is a simple test to see what workers are available in your cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['sparky-w-1-79n6p']"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import socket\n",
    "spark.range(100, numPartitions=100).rdd.map(lambda x: socket.gethostname()).distinct().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Read a simple text file from S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "df0 = spark.read.text(\"s3a://penasio/spark-test\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'fields': [{'metadata': {},\n",
       "   'name': 'value',\n",
       "   'nullable': True,\n",
       "   'type': 'string'}],\n",
       " 'type': 'struct'}"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df0.schema.jsonValue()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|               value|\n",
      "+--------------------+\n",
      "|      # Apache Spark|\n",
      "|                    |\n",
      "|Spark is a fast a...|\n",
      "|high-level APIs i...|\n",
      "|supports general ...|\n",
      "|rich set of highe...|\n",
      "|MLlib for machine...|\n",
      "|and Spark Streami...|\n",
      "|                    |\n",
      "|<http://spark.apa...|\n",
      "+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df0.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u' ', 464), (u'e', 270), (u'a', 253), (u't', 219), (u'o', 212)]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from operator import add\n",
    "df0.rdd.flatMap(lambda x: list(x[0])).map(lambda x: (x, 1)).reduceByKey(add).sortBy(lambda x: x[1], ascending=False).take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Word count example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'', 72),\n",
       " (u'project.', 1),\n",
       " (u'help', 1),\n",
       " (u'when', 1),\n",
       " (u'Hadoop', 3),\n",
       " (u'not', 1),\n",
       " (u'./dev/run-tests', 1),\n",
       " (u'including', 4),\n",
       " (u'graph', 1),\n",
       " (u'computation', 1)]"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "text_file = spark.read.text(\"s3a://penasio/spark-test\")\n",
    "counts = text_file.rdd.flatMap(lambda line: line[0].split(\" \")) \\\n",
    "             .map(lambda word: (word, 1)) \\\n",
    "             .reduceByKey(add) \\\n",
    "             .take(10)\n",
    "counts"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "celltoolbar": "Raw Cell Format",
  "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}